83 lines
2.8 KiB
Python
83 lines
2.8 KiB
Python
import sys
|
|
import datetime
|
|
|
|
from sqlalchemy import BigInteger
|
|
from sqlalchemy.dialects import mysql, sqlite
|
|
from sqlalchemy.types import DateTime, TypeDecorator
|
|
|
|
|
|
class ModelSerializeMixin:
|
|
"""Mixin class used for models to serialize them automatically
|
|
Ignores private and protected members as well as members marked as not to publish (name ends with _)
|
|
"""
|
|
|
|
def __is_optional(self, param):
|
|
if sys.version_info < (3, 8):
|
|
return False
|
|
|
|
import typing
|
|
|
|
hint = typing.get_type_hints(self.__class__)[param]
|
|
if (
|
|
typing.get_origin(hint) is typing.Union
|
|
and len(typing.get_args(hint)) == 2
|
|
and typing.get_args(hint)[1] is type(None)
|
|
):
|
|
return getattr(self, param) is None
|
|
|
|
def serialize(self):
|
|
"""Serialize class to dict
|
|
Returns:
|
|
Dict of all not private or protected annotated member variables.
|
|
"""
|
|
d = {
|
|
param: getattr(self, param)
|
|
for param in self.__class__.__annotations__
|
|
if not param.startswith("_") and not param.endswith("_") and not self.__is_optional(param)
|
|
}
|
|
if len(d) == 1:
|
|
key, value = d.popitem()
|
|
return value
|
|
return d
|
|
|
|
|
|
class Serial(TypeDecorator):
|
|
"""Same as MariaDB Serial used for IDs"""
|
|
|
|
impl = BigInteger().with_variant(mysql.BIGINT(unsigned=True), "mysql").with_variant(sqlite.INTEGER, "sqlite")
|
|
|
|
|
|
class UtcDateTime(TypeDecorator):
|
|
"""Almost equivalent to `sqlalchemy.types.DateTime` with
|
|
``timezone=True`` option, but it differs from that by:
|
|
|
|
- Never silently take naive :class:`datetime.datetime`, instead it
|
|
always raise :exc:`ValueError` unless time zone aware value.
|
|
- :class:`datetime.datetime` value's :attr:`datetime.datetime.tzinfo`
|
|
is always converted to UTC.
|
|
- Unlike SQLAlchemy's built-in :class:`sqlalchemy.types.DateTime`,
|
|
it never return naive :class:`datetime.datetime`, but time zone
|
|
aware value, even with SQLite or MySQL.
|
|
"""
|
|
|
|
impl = DateTime(timezone=True)
|
|
|
|
@staticmethod
|
|
def current_utc():
|
|
return datetime.datetime.now(tz=datetime.timezone.utc)
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
if value is not None:
|
|
if not isinstance(value, datetime.datetime):
|
|
raise TypeError("expected datetime.datetime, not " + repr(value))
|
|
elif value.tzinfo is None:
|
|
raise ValueError("naive datetime is disallowed")
|
|
return value.astimezone(datetime.timezone.utc)
|
|
|
|
def process_result_value(self, value, dialect):
|
|
if value is not None:
|
|
if value.tzinfo is not None:
|
|
value = value.astimezone(datetime.timezone.utc)
|
|
value = value.replace(tzinfo=datetime.timezone.utc)
|
|
return value
|