""" DBServiceConfig - Configuration for DB Service permissions and filtering. Allows application layer to register model scopes and column names, making the DB service completely generic and plug-and-play. """ import logging from typing import Type, Set, Optional logger = logging.getLogger(__name__) class DBServiceConfig: """ Centralized configuration for DB Service. Register your application's models and permissions at startup. """ # Configuration state _registered = False # Database metadata db_base = None # SQLAlchemy declarative base all_models: list = [] # All model classes # Column names (generic) user_filter_column: str = "user_id" user_id_column: str = "id" soft_delete_column: str = "deleted_at" # Special models special_user_model: Optional[Type] = None # USER scopes user_read_scoped: Set[Type] = set() user_create_scoped: Set[Type] = set() user_update_scoped: Set[Type] = set() user_delete_scoped: Set[Type] = set() # ADMIN scopes admin_read_only: Set[Type] = set() admin_create_only: Set[Type] = set() admin_update_only: Set[Type] = set() admin_delete_only: Set[Type] = set() # SYSTEM scopes system_read_scoped: Set[Type] = set() system_create_scoped: Set[Type] = set() system_update_scoped: Set[Type] = set() system_delete_scoped: Set[Type] = set() @classmethod def register( cls, # Database metadata db_base=None, # SQLAlchemy Base (for table creation) all_models: list = None, # All model classes # Column names user_filter_column: str = "user_id", user_id_column: str = "id", soft_delete_column: str = "deleted_at", # Special models special_user_model: Optional[Type] = None, # USER scopes user_read_scoped: list = None, user_create_scoped: list = None, user_update_scoped: list = None, user_delete_scoped: list = None, # ADMIN scopes admin_read_only: list = None, admin_create_only: list = None, admin_update_only: list = None, admin_delete_only: list = None, # SYSTEM scopes system_read_scoped: list = None, system_create_scoped: list = None, system_update_scoped: list = None, system_delete_scoped: list = None, ) -> None: """Register DB Service configuration at application startup.""" # Database metadata cls.db_base = db_base cls.all_models = all_models or [] # Column names cls.user_filter_column = user_filter_column cls.user_id_column = user_id_column cls.soft_delete_column = soft_delete_column # Special models cls.special_user_model = special_user_model # USER scopes cls.user_read_scoped = set(user_read_scoped or []) cls.user_create_scoped = set(user_create_scoped or []) cls.user_update_scoped = set(user_update_scoped or []) cls.user_delete_scoped = set(user_delete_scoped or []) # ADMIN scopes cls.admin_read_only = set(admin_read_only or []) cls.admin_create_only = set(admin_create_only or []) cls.admin_update_only = set(admin_update_only or []) cls.admin_delete_only = set(admin_delete_only or []) # SYSTEM scopes cls.system_read_scoped = set(system_read_scoped or []) cls.system_create_scoped = set(system_create_scoped or []) cls.system_update_scoped = set(system_update_scoped or []) cls.system_delete_scoped = set(system_delete_scoped or []) cls._registered = True logger.info("✅ DBServiceConfig registered successfully") logger.info(f" Models registered: {len(cls.all_models)}") logger.info(f" User filter column: {cls.user_filter_column}") logger.info(f" Soft delete column: {cls.soft_delete_column}") logger.info(f" USER scopes: {len(cls.user_read_scoped)} read, {len(cls.user_create_scoped)} create") logger.info(f" ADMIN scopes: {len(cls.admin_read_only)} read, {len(cls.admin_create_only)} create") logger.info(f" SYSTEM scopes: {len(cls.system_read_scoped)} read, {len(cls.system_create_scoped)} create") @classmethod def is_registered(cls) -> bool: """Check if configuration has been registered.""" return cls._registered @classmethod def assert_registered(cls) -> None: """Assert that configuration has been registered, raise if not.""" if not cls._registered: raise RuntimeError( "DBServiceConfig not registered! " "Call DBServiceConfig.register() at application startup." )