"""Utilities for common models.""" from datetime import datetime from typing import Any, TypeVar from sqlalchemy import inspect from sqlalchemy.schema import Column from . import constants from .errors import ( ERROR_ENTITY_COMPOSITE_KEY_NOT_SUPPORTED, ERROR_NOT_CREATABLE, ERROR_NOT_SOFT_DELETABLE, ERROR_NOT_UPDATABLE, ) T = TypeVar('T') def current_timestamp() -> datetime: """Get the current timestamp in SYSTEM_TIMEZONE as a naive datetime (no tzinfo).""" return datetime.now(tz=constants.SYSTEM_TIMEZONE).replace(tzinfo=None) def get_class(value: T | type[T]) -> type[T]: """Get the class of a value. Args: value: Class or class instance. Returns: The class """ return value if isinstance(value, type) else type(value) def get_pk(instance: Any) -> dict[str, Any]: """Return a dict of primary keys and values.""" cols = get_pk_columns(instance) return {col.name: getattr(instance, col.name, None) for col in cols} def get_pk_columns(cls: Any | type[Any]) -> list[Column[Any]]: """Return primary key columns. Args: cls: Model class or instance to get primary key columns from. Returns: List of primary key Column objects. """ # Handle both instance and class mapper = inspect(get_class(cls)) return mapper.primary_key def is_creatable(model: Any) -> bool: """Check if model supports creation fields. A model is creatable if it has either `created_at` or `created_by`. """ return hasattr(model, 'created_at') or hasattr(model, 'created_by') def is_soft_deletable(model: Any) -> bool: """Check if model supports soft deletion. A model is soft-deletable if it has either `deleted_at` or `deleted_by`. """ return hasattr(model, 'deleted_at') or hasattr(model, 'deleted_by') def is_updatable(model: Any) -> bool: """Check if model supports update tracking. A model is updatable if it has either `last_modified` or `last_modified_by`. """ return hasattr(model, 'last_modified') or hasattr(model, 'last_modified_by') def remove_from_registry(cls: type[Any]) -> None: """Clear a mapping from the SQLAlchemy registry. Args: cls: SQLAlchemy model class to remove from the registry """ try: # Get the base registry base_registry = cls.registry # Get the class registry class_registry = base_registry._class_registry # Clear the simple name class_registry.pop(cls.__name__, None) except (AttributeError, KeyError): # Registry might not exist or class not registered pass def try_creatable(model: Any) -> None: """Raise error if model is not creatable.""" if not is_creatable(model): raise TypeError(ERROR_NOT_CREATABLE) def try_pk_column(instance: Any) -> Column[Any]: """Return single primary key value or raise error.""" # Handle both class and instance cls = get_class(instance) cols = get_pk_columns(cls) if len(cols) == 1: return cols[0] raise ValueError( ERROR_ENTITY_COMPOSITE_KEY_NOT_SUPPORTED.format( object_type=cls.__name__, num_cols=len(cols) ) ) def try_soft_deletable(model: Any) -> None: """Raise error if model is not soft-deletable.""" if not is_soft_deletable(model): raise TypeError(ERROR_NOT_SOFT_DELETABLE) def try_updatable(model: Any) -> None: """Raise error if model is not updatable.""" if not is_updatable(model): raise TypeError(ERROR_NOT_UPDATABLE)