80 lines
2.9 KiB
Python
80 lines
2.9 KiB
Python
from core.models.Model import Model
|
|
from dataclasses import dataclass, field
|
|
from dataclasses_json import config, dataclass_json, Exclude
|
|
from datetime import datetime
|
|
from dateutil.parser import isoparse
|
|
from marshmallow import fields
|
|
from typing import Optional
|
|
|
|
_table_name: str = 'client_versions'
|
|
|
|
_table_definition: str = """
|
|
'id' int UNIQUE,
|
|
'version_number' varchar UNIQUE,
|
|
'released_at' varchar,
|
|
'download_path' varchar UNIQUE
|
|
"""
|
|
|
|
|
|
@dataclass_json
|
|
@dataclass
|
|
class ClientVersion(Model):
|
|
version_number: str
|
|
released_at: datetime = field(
|
|
metadata=config(
|
|
encoder=datetime.isoformat,
|
|
decoder=datetime.fromisoformat,
|
|
mm_field=fields.DateTime(format='iso')
|
|
)
|
|
)
|
|
id: Optional[int] = field(
|
|
default=None,
|
|
metadata=config(exclude=Exclude.ALWAYS)
|
|
)
|
|
download_path: Optional[str] = field(
|
|
default=None,
|
|
metadata=config(exclude=Exclude.ALWAYS)
|
|
)
|
|
|
|
@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 client_versions WHERE id = ? LIMIT 1', ClientVersion.factory, [id])
|
|
|
|
@staticmethod
|
|
def find(version_number: str):
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
return Model._query_one('SELECT * FROM client_versions WHERE version_number = ? LIMIT 1', ClientVersion.factory, [version_number])
|
|
|
|
@staticmethod
|
|
def latest():
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
return Model._query_one('SELECT * FROM client_versions ORDER BY CAST(released_at as datetime) DESC LIMIT 1', ClientVersion.factory)
|
|
|
|
@staticmethod
|
|
def all():
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
return Model._query_all('SELECT * FROM client_versions', ClientVersion.factory)
|
|
|
|
@staticmethod
|
|
def truncate():
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition, drop_existing=True)
|
|
|
|
@staticmethod
|
|
def save_many(client_versions):
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
Model._insert_many('INSERT INTO client_versions VALUES(?, ?, ?, ?)', ClientVersion.tuple_factory, client_versions)
|
|
|
|
@staticmethod
|
|
def factory(cursor, row):
|
|
|
|
database_fields = [column[0] for column in cursor.description]
|
|
|
|
client_version = ClientVersion(**{key: value for key, value in zip(database_fields, row)})
|
|
client_version.released_at = isoparse(str(client_version.released_at))
|
|
|
|
return client_version
|
|
|
|
@staticmethod
|
|
def tuple_factory(client_version):
|
|
return client_version.id, client_version.version_number, client_version.released_at, client_version.download_path
|