|
|
# src/common/models.py
|
|
|
import uuid
|
|
|
from datetime import datetime
|
|
|
from typing import Generic, Optional, TypeVar
|
|
|
|
|
|
import pytz
|
|
|
from pydantic import BaseModel, Field, SecretStr,Extra
|
|
|
from sqlalchemy import Column, DateTime, String, event, func
|
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
|
|
from src.auth.service import CurrentUser
|
|
|
from src.config import TIMEZONE
|
|
|
from src.enums import ResponseStatus
|
|
|
|
|
|
# SQLAlchemy Mixins
|
|
|
|
|
|
|
|
|
class TimeStampMixin(object):
|
|
|
"""Timestamping mixin"""
|
|
|
|
|
|
created_at = Column(
|
|
|
DateTime(timezone=True), default=datetime.now(pytz.timezone(TIMEZONE))
|
|
|
)
|
|
|
created_at._creation_order = 9998
|
|
|
updated_at = Column(
|
|
|
DateTime(timezone=True), default=datetime.now(pytz.timezone(TIMEZONE))
|
|
|
)
|
|
|
updated_at._creation_order = 9998
|
|
|
|
|
|
@staticmethod
|
|
|
def _updated_at(mapper, connection, target):
|
|
|
target.updated_at = datetime.now(pytz.timezone(TIMEZONE))
|
|
|
|
|
|
@classmethod
|
|
|
def __declare_last__(cls):
|
|
|
event.listen(cls, "before_update", cls._updated_at)
|
|
|
|
|
|
|
|
|
class UUIDMixin:
|
|
|
"""UUID mixin"""
|
|
|
|
|
|
id = Column(
|
|
|
UUID(as_uuid=True),
|
|
|
primary_key=True,
|
|
|
default=uuid.uuid4,
|
|
|
unique=True,
|
|
|
nullable=False,
|
|
|
)
|
|
|
|
|
|
|
|
|
class DefaultMixin(TimeStampMixin, UUIDMixin):
|
|
|
"""Default mixin"""
|
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
class IdentityMixin:
|
|
|
"""Identity mixin"""
|
|
|
|
|
|
created_by = Column(String(100), nullable=True)
|
|
|
updated_by = Column(String(100), nullable=True)
|
|
|
|
|
|
|
|
|
# Pydantic Models
|
|
|
class DefultBase(BaseModel):
|
|
|
class Config:
|
|
|
# allow model creation directly from ORM objects or attrs
|
|
|
from_attributes = True
|
|
|
|
|
|
# re-validate fields when they’re updated after creation
|
|
|
validate_assignment = True
|
|
|
|
|
|
# disallow arbitrary/untyped objects; only safe, supported types allowed
|
|
|
arbitrary_types_allowed = False
|
|
|
|
|
|
# automatically strip leading/trailing whitespace from all str fields
|
|
|
str_strip_whitespace = True
|
|
|
|
|
|
# forbid extra/unexpected fields in input (prevents silent injection/mass assignment)
|
|
|
extra = 'forbid'
|
|
|
|
|
|
# secure JSON serialization: custom formatting for sensitive types
|
|
|
json_encoders = {
|
|
|
# always output datetime in strict ISO8601 with Zulu timezone
|
|
|
datetime: lambda v: v.strftime("%Y-%m-%dT%H:%M:%S.%fZ") if v else None,
|
|
|
# unwrap SecretStr safely when serializing (not leaking in repr/debug)
|
|
|
SecretStr: lambda v: v.get_secret_value() if v else None,
|
|
|
}
|
|
|
|
|
|
|
|
|
class Pagination(DefultBase):
|
|
|
itemsPerPage: int
|
|
|
totalPages: int
|
|
|
page: int
|
|
|
total: int
|
|
|
|
|
|
|
|
|
class PrimaryKeyModel(BaseModel):
|
|
|
id: uuid.UUID
|
|
|
|
|
|
|
|
|
# Define data type variable for generic response
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
|
|
|
class StandardResponse(BaseModel, Generic[T]):
|
|
|
data: Optional[T] = None
|
|
|
message: str = "Success"
|
|
|
status: ResponseStatus = ResponseStatus.SUCCESS
|