Spaces:
Sleeping
Sleeping
File size: 4,884 Bytes
50c20bf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 |
"""
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."
)
|