134 lines
5.1 KiB
Python
134 lines
5.1 KiB
Python
from core.Constants import Constants
|
|
from core.models.Model import Model
|
|
from core.models.session.Application import Application
|
|
from dataclasses import dataclass, field
|
|
from dataclasses_json import config, Exclude
|
|
from datetime import datetime
|
|
from dateutil.parser import isoparse
|
|
from marshmallow import fields
|
|
from typing import Optional
|
|
import os
|
|
|
|
_table_name: str = 'application_versions'
|
|
|
|
_table_definition: str = """
|
|
'id' int UNIQUE,
|
|
'application_code' varchar,
|
|
'version_number' varchar,
|
|
'format_revision' int,
|
|
'download_path' varchar UNIQUE,
|
|
'released_at' varchar,
|
|
'file_hash' varchar,
|
|
UNIQUE(application_code, version_number)
|
|
"""
|
|
|
|
|
|
@dataclass
|
|
class ApplicationVersion(Model):
|
|
application_code: str
|
|
version_number: str
|
|
format_revision: Optional[int] = field(
|
|
default=None,
|
|
metadata=config(exclude=Exclude.ALWAYS)
|
|
)
|
|
id: Optional[int] = field(
|
|
default=None,
|
|
metadata=config(exclude=Exclude.ALWAYS)
|
|
)
|
|
download_path: Optional[str] = field(
|
|
default=None,
|
|
metadata=config(exclude=Exclude.ALWAYS)
|
|
)
|
|
released_at: Optional[datetime] = field(
|
|
default=None,
|
|
metadata=config(
|
|
encoder=datetime.isoformat,
|
|
decoder=datetime.fromisoformat,
|
|
mm_field=fields.DateTime(format='iso'),
|
|
exclude=Exclude.ALWAYS
|
|
)
|
|
)
|
|
file_hash: Optional[str] = field(
|
|
default=None,
|
|
metadata=config(exclude=Exclude.ALWAYS)
|
|
)
|
|
installed: Optional[bool] = field(
|
|
default=False,
|
|
metadata=config(exclude=Exclude.ALWAYS)
|
|
)
|
|
supported: Optional[bool] = field(
|
|
default=False,
|
|
metadata=config(exclude=Exclude.ALWAYS)
|
|
)
|
|
|
|
def __post_init__(self):
|
|
self.installed = self.is_installed()
|
|
self.supported = self.is_supported()
|
|
|
|
def get_installation_path(self):
|
|
return f'{Constants.HV_APPLICATION_DATA_HOME}/{self.application_code}/{self.version_number}'
|
|
|
|
def is_installed(self):
|
|
return os.path.isdir(self.get_installation_path()) and len(os.listdir(self.get_installation_path())) > 0
|
|
|
|
def is_supported(self):
|
|
return self.exists(self.application_code, self.version_number) and self.format_revision == 1
|
|
|
|
def get_installed_file_hash(self):
|
|
|
|
try:
|
|
return open(f'{self.get_installation_path()}/.sha3-512').readline().strip()
|
|
except FileNotFoundError:
|
|
return None
|
|
|
|
def is_fresh(self):
|
|
return self.is_installed() and (not self.is_supported() or self.file_hash == self.get_installed_file_hash())
|
|
|
|
@staticmethod
|
|
def find_by_id(id: int):
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
return Model._query_one('SELECT * FROM application_versions WHERE id = ? LIMIT 1', ApplicationVersion.factory, [id])
|
|
|
|
@staticmethod
|
|
def find(application_code: str, version_number: str):
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
return Model._query_one('SELECT * FROM application_versions WHERE application_code = ? AND version_number = ? LIMIT 1', ApplicationVersion.factory, [application_code, version_number])
|
|
|
|
@staticmethod
|
|
def all(application: Optional[Application] = None):
|
|
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
|
|
if application is None:
|
|
return Model._query_all('SELECT * FROM application_versions', ApplicationVersion.factory)
|
|
|
|
else:
|
|
return Model._query_all('SELECT * FROM application_versions WHERE application_code = ?', ApplicationVersion.factory, [application.code])
|
|
|
|
@staticmethod
|
|
def exists(application_code: str, version_number: str):
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
return Model._query_exists('SELECT * FROM application_versions WHERE application_code = ? AND version_number = ?', [application_code, version_number])
|
|
|
|
@staticmethod
|
|
def truncate():
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition, drop_existing=True)
|
|
|
|
@staticmethod
|
|
def save_many(application_versions):
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
Model._insert_many('INSERT INTO application_versions VALUES(?, ?, ?, ?, ?, ?, ?)', ApplicationVersion.tuple_factory, application_versions)
|
|
|
|
@staticmethod
|
|
def factory(cursor, row):
|
|
|
|
database_fields = [column[0] for column in cursor.description]
|
|
|
|
application_version = ApplicationVersion(**{key: value for key, value in zip(database_fields, row)})
|
|
application_version.released_at = isoparse(str(application_version.released_at))
|
|
|
|
return application_version
|
|
|
|
@staticmethod
|
|
def tuple_factory(application_version):
|
|
return application_version.id, application_version.application_code, application_version.version_number, application_version.format_revision, application_version.download_path, application_version.released_at, application_version.file_hash
|