94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
from datetime import datetime
|
|
from typing import NewType
|
|
from sqlalchemy import MetaData, Uuid, Index
|
|
from sqlalchemy_declarative_extensions import declarative_database
|
|
from sqlalchemy_declarative_extensions.audit import audit
|
|
from sqlalchemy.orm import declarative_base, declared_attr, Mapped, mapped_column
|
|
from sqlalchemy.dialects.postgresql import INET, JSONB, ARRAY
|
|
import uuid
|
|
|
|
nameing_metadata = MetaData(naming_convention= {
|
|
"ix": "ix_%(column_0_label)s",
|
|
"uq": "uq_%(table_name)s_%(column_0_name)s",
|
|
"ck": "ck_%(table_name)s_%(constraint_name)s",
|
|
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
|
|
"pk": "pk_%(table_name)s",
|
|
})
|
|
|
|
nstr30 = NewType("nstr30", str)
|
|
nstr50 = NewType("nstr50", str)
|
|
nstr100 = NewType("nstr100", str)
|
|
nstr255 = NewType("nstr255", str)
|
|
tz_date = NewType("tz_date", datetime)
|
|
|
|
Base = declarative_database(declarative_base(metadata=nameing_metadata))
|
|
|
|
|
|
|
|
class ApiLog(Base):
|
|
__tablename__ = 'api_log'
|
|
|
|
id : Mapped[uuid.UUID] = mapped_column(primary_key=True, default = uuid.uuid7)
|
|
api_key : Mapped[uuid.UUID | None]
|
|
ip_address : Mapped[str] = mapped_column(INET)
|
|
path : Mapped[str]
|
|
method : Mapped[str]
|
|
status_code : Mapped[int]
|
|
request_body : Mapped[dict | None] = mapped_column(JSONB)
|
|
response_body : Mapped[dict | None] = mapped_column(JSONB)
|
|
query_params : Mapped[dict | None] = mapped_column(JSONB)
|
|
path_params : Mapped[dict | None] = mapped_column(JSONB)
|
|
process_time : Mapped[float]
|
|
created_at : Mapped[datetime]
|
|
|
|
class AuditMixin:
|
|
@declared_attr
|
|
def created_at(cls) -> Mapped[tz_date]:
|
|
return mapped_column()
|
|
|
|
@declared_attr
|
|
def updated_at(cls) -> Mapped[tz_date]:
|
|
return mapped_column()
|
|
|
|
@declared_attr
|
|
def created_by(cls) -> Mapped[Users | None]:
|
|
return mapped_column(foreign_key=[cls.created_by_id], remote_side=[Users.id])
|
|
|
|
@declared_attr
|
|
def updated_by(cls) -> Mapped[Users | None]:
|
|
return mapped_column(foreign_key=[cls.updated_by_id], remote_side=[Users.id])
|
|
|
|
@declared_attr
|
|
def updated_by_id(cls) -> Mapped[Uuid]:
|
|
return mapped_column()
|
|
|
|
@declared_attr
|
|
def created_by_id(cls) -> Mapped[Uuid]:
|
|
return mapped_column()
|
|
|
|
|
|
@audit()
|
|
class Users(Base):
|
|
__tablename__ = "users"
|
|
id : Mapped[Uuid] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
|
|
username : Mapped[nstr50] = mapped_column()
|
|
full_name : Mapped[nstr100]
|
|
email : Mapped[nstr255] = mapped_column()
|
|
type : Mapped[nstr50] = mapped_column(default="staff")
|
|
is_archived : Mapped[bool] = mapped_column(default=False)
|
|
hashed_password: Mapped[nstr255]
|
|
|
|
__table_args__ = (
|
|
Index("ix_unique_active_email", email, unique=True, postgresql_where="is_archived = false"),
|
|
Index("ix_unique_active_username", username, unique=True, postgresql_where="is_archived = false"),
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|