generated from daniil-berg/boilerplate-py
36 lines
1.4 KiB
Python
36 lines
1.4 KiB
Python
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
|
from sqlalchemy.orm import declarative_base, sessionmaker
|
|
from sqlalchemy.sql.functions import now as db_now
|
|
from sqlalchemy.sql.schema import Column
|
|
from sqlalchemy.sql.sqltypes import TIMESTAMP
|
|
from sqlalchemy_utils.functions.orm import get_columns
|
|
from sqlalchemy_utils.listeners import force_auto_coercion
|
|
|
|
from compub.exceptions import NoDatabaseConfigured
|
|
from compub.settings import settings
|
|
|
|
|
|
if settings.db_uri is None:
|
|
raise NoDatabaseConfigured
|
|
engine = create_async_engine(settings.db_uri, future=True)
|
|
LocalAsyncSession = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
|
|
ORMBase = declarative_base()
|
|
|
|
force_auto_coercion()
|
|
|
|
|
|
class AbstractBase(ORMBase):
|
|
__abstract__ = True
|
|
|
|
NON_REPR_FIELDS = ['id', 'date_created', 'date_updated']
|
|
|
|
date_created = Column(TIMESTAMP(timezone=False), server_default=db_now())
|
|
date_updated = Column(TIMESTAMP(timezone=False), server_default=db_now(), onupdate=db_now())
|
|
|
|
def __repr__(self) -> str:
|
|
# Exclude non-representative fields:
|
|
fields = (name for name in get_columns(self).keys() if name not in self.NON_REPR_FIELDS)
|
|
# Exclude NULL value fields:
|
|
attrs = ', '.join(f"{name}={repr(getattr(self, name))}" for name in fields if getattr(self, name) is not None)
|
|
return f"{self.__class__.__name__}({attrs})"
|