from sqlalchemy import ( TIMESTAMP, NUMERIC, func, Identity, UUID, String, Integer, Boolean, SmallInteger, ) from sqlalchemy.orm import ( Mapped, mapped_column, InstrumentedAttribute, ) from sqlalchemy_mixins.session import SessionMixin from sqlalchemy_mixins.serialize import SerializeMixin from sqlalchemy_mixins.repr import ReprMixin from sqlalchemy_mixins.smartquery import SmartQueryMixin from api_library.date_time_actions.date_functions import DateTimeLocal from api_objects.auth.token_objects import Credentials from databases.sql_models.sql_operations import FilterAttributes from databases.sql_models.postgres_database import Base class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes): __abstract__ = True # The model is abstract not a database table. __session__ = Base.session # The session to use in the model. __system__fields__create__ = ( "ref_id", "created_at", "updated_at", "cryp_uu_id", "created_by", "created_by_id", "updated_by", "updated_by_id", "replication_id", "confirmed_by", "confirmed_by_id", "is_confirmed", "deleted", "active", "is_notification_send", "is_email_send", "expiry_starts", "expiry_ends", ) # The system fields to use in the model. __system__fields__update__ = ( "cryp_uu_id", "created_at", "updated_at", "created_by", "created_by_id", "confirmed_by", "confirmed_by_id", "updated_by", "updated_by_id", "replication_id", ) creds: Credentials = None # The credentials to use in the model. client_arrow: DateTimeLocal = None # The arrow to use in the model. expiry_starts: Mapped[TIMESTAMP] = mapped_column( TIMESTAMP, server_default=func.now(), nullable=False ) expiry_ends: Mapped[TIMESTAMP] = mapped_column( TIMESTAMP, default="2099-12-31", server_default="2099-12-31" ) @classmethod def extract_system_fields(cls, filter_kwargs: dict, create: bool = True): """ Extracts the system fields from the given attributes. """ system_fields = filter_kwargs.copy() extract_fields = ( cls.__system__fields__create__ if create else cls.__system__fields__update__ ) for field in extract_fields: system_fields.pop(field, None) return system_fields @classmethod def find_or_create(cls, **kwargs): from api_library.date_time_actions.date_functions import system_arrow """ Finds a record with the given attributes or creates it if it doesn't exist. If found, sets is_found to True, otherwise False. is_found can be used to check if the record was found or created. """ check_kwargs = cls.extract_system_fields(kwargs) cls.pre_query = cls.query.filter( system_arrow.get(cls.expiry_ends).date() < system_arrow.now().date() ) already_record = cls.filter_by_one(**check_kwargs) if already_record := already_record.data: if already_record.is_deleted: cls.raise_http_exception( status_code="HTTP_406_NOT_ACCEPTABLE", error_case="DeletedRecord", data=check_kwargs, message="Record exits but is deleted. Contact with authorized user", ) elif already_record.is_confirmed: cls.raise_http_exception( status_code="HTTP_406_NOT_ACCEPTABLE", error_case="IsNotConfirmed", data=check_kwargs, message="Record exits but is not confirmed. Contact with authorized user", ) cls.raise_http_exception( status_code="HTTP_406_NOT_ACCEPTABLE", error_case="AlreadyExists", data=check_kwargs, message="Record already exits. Refresh data and try again", ) created_record = cls() for key, value in check_kwargs.items(): setattr(created_record, key, value) created_record.flush() cls.created_by_id = cls.creds.person_id cls.created_by = cls.creds.person_name return created_record @classmethod def iterate_over_variables(cls, val, key): key_ = cls.__annotations__.get(key, None) if val is None or key_ is None: return None elif key_ == Mapped[Identity]: return None elif key_ == Mapped[bool]: return bool(val) elif key_ == Mapped[float] or key_ == Mapped[NUMERIC]: return float(val) elif key_ == Mapped[int]: return int(val) elif key_ == Mapped[TIMESTAMP]: return str(cls.client_arrow.get(val).format("DD-MM-YYYY HH:mm:ss")) return str(val) def update(self, **kwargs): """Updates the record with the given attributes.""" is_confirmed_argument = kwargs.get("is_confirmed", None) if is_confirmed_argument and not len(kwargs) == 1: self.raise_http_exception( status_code="HTTP_406_NOT_ACCEPTABLE", error_case="ConfirmError", data=kwargs, message="Confirm field can not be updated with other fields", ) check_kwargs = self.extract_system_fields(kwargs, create=False) for key, value in check_kwargs.items(): setattr(self, key, value) if is_confirmed_argument: self.confirmed_by_id = self.creds.person_id self.confirmed_by = self.creds.person_name else: self.updated_by_id = self.creds.person_id self.updated_by = self.creds.person_name self.flush() return self def get_dict( self, exclude: list = None, include: list = None, include_joins: list = None ): return_dict = {} if exclude: exclude.extend(list(set(self.__exclude__fields__).difference(exclude))) else: exclude = self.__exclude__fields__ include = include or [] if include: include.extend(["uu_id", "active"]) include = list(set(include).difference(self.__exclude__fields__)) for key, val in self.to_dict().items(): if key in include: if value_of_database := self.iterate_over_variables(val, key): return_dict[key] = value_of_database else: exclude.extend(["is_confirmed", "deleted", "cryp_uu_id"]) for key, val in self.to_dict().items(): if key not in exclude: if value_of_database := self.iterate_over_variables(val, key): return_dict[key] = value_of_database all_arguments = [ record for record in self.__class__.__dict__ if "_" not in record[0] and "id" not in record[-2:] ] for all_argument in all_arguments: column = getattr(self.__class__, all_argument) is_populate = isinstance(column, InstrumentedAttribute) and not hasattr( column, "foreign_keys" ) if is_populate and all_argument in include_joins or []: populate_arg = getattr(self, all_argument, None) if isinstance(populate_arg, list): return_dict[all_argument] = [ arg.get_dict() if arg else [] for arg in populate_arg ] elif getattr(populate_arg, "get_dict", None): return_dict[all_argument] = ( populate_arg.get_dict() if populate_arg else [] ) return return_dict class BaseMixin(CrudMixin, ReprMixin, SerializeMixin): __abstract__ = True class BaseCollection(CrudMixin, BaseMixin): __abstract__ = True __repr__ = ReprMixin.__repr__ id: Mapped[Identity] = mapped_column(primary_key=True) class CrudCollection(CrudMixin, BaseMixin, SmartQueryMixin): __abstract__ = True __repr__ = ReprMixin.__repr__ id: Mapped[Identity] = mapped_column(primary_key=True) uu_id: Mapped[UUID] = mapped_column( UUID, server_default=func.text("gen_random_uuid()"), index=True, unique=True ) ref_id: Mapped[UUID] = mapped_column(String(100), nullable=True, index=True) created_at: Mapped[TIMESTAMP] = mapped_column( "created_at", TIMESTAMP(timezone=True), server_default=func.now(), nullable=False, index=True, ) updated_at: Mapped[TIMESTAMP] = mapped_column( "updated_at", TIMESTAMP(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False, index=True, ) cryp_uu_id: Mapped[UUID] = mapped_column(String, nullable=True, index=True) created_by: Mapped[str] = mapped_column(String, nullable=True) created_by_id: Mapped[int] = mapped_column(Integer, nullable=True) updated_by: Mapped[str] = mapped_column(String, nullable=True) updated_by_id: Mapped[int] = mapped_column(Integer, nullable=True) confirmed_by: Mapped[str] = mapped_column(String, nullable=True) confirmed_by_id: Mapped[int] = mapped_column(Integer, nullable=True) is_confirmed: Mapped[bool] = mapped_column(Boolean, server_default="0") replication_id: Mapped[int] = mapped_column(SmallInteger, server_default="0") deleted: Mapped[bool] = mapped_column(Boolean, server_default="0") active: Mapped[bool] = mapped_column(Boolean, server_default="1") is_notification_send: Mapped[bool] = mapped_column(Boolean, server_default="0") is_email_send: Mapped[bool] = mapped_column(Boolean, server_default="0")