You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
86 lines
2.2 KiB
Python
86 lines
2.2 KiB
Python
# src/common/models.py
|
|
from datetime import datetime
|
|
from typing import Generic, Optional, TypeVar
|
|
import uuid
|
|
from pydantic import BaseModel, Field, SecretStr
|
|
from sqlalchemy import Column, DateTime, String, func, event
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
from src.config import TIMEZONE
|
|
import pytz
|
|
|
|
from src.auth.service import CurrentUser
|
|
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 IdentityMixin:
|
|
"""Identity mixin"""
|
|
created_by = Column(String(100), nullable=True)
|
|
updated_by = Column(String(100), nullable=True)
|
|
|
|
|
|
class DefaultMixin(TimeStampMixin, UUIDMixin):
|
|
"""Default mixin"""
|
|
pass
|
|
|
|
|
|
# Pydantic Models
|
|
class DefultBase(BaseModel):
|
|
class Config:
|
|
from_attributes = True
|
|
validate_assignment = True
|
|
arbitrary_types_allowed = True
|
|
str_strip_whitespace = True
|
|
|
|
json_encoders = {
|
|
# custom output conversion for datetime
|
|
datetime: lambda v: v.strftime("%Y-%m-%dT%H:%M:%S.%fZ") if v else None,
|
|
SecretStr: lambda v: v.get_secret_value() if v else None,
|
|
}
|
|
|
|
|
|
class Pagination(DefultBase):
|
|
itemsPerPage: 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
|