# [DEF:AuthRepositoryModule:Module] # @TIER: CRITICAL # @COMPLEXITY: 5 # @SEMANTICS: auth, repository, database, user, role, permission # @PURPOSE: Data access layer for authentication and user preference entities. # @LAYER: Domain # @RELATION: DEPENDS_ON -> [Session] # @RELATION: DEPENDS_ON -> [User] # @RELATION: DEPENDS_ON -> [Role] # @RELATION: DEPENDS_ON -> [Permission] # @RELATION: DEPENDS_ON -> [UserDashboardPreference] # @RELATION: DEPENDS_ON -> [belief_scope] # @INVARIANT: All database read/write operations must execute via the injected SQLAlchemy session boundary. # @DATA_CONTRACT: Session -> [User | Role | Permission | UserDashboardPreference] # @PRE: Database connection is active. # @POST: Provides valid access to identity data. # @SIDE_EFFECT: None at module level. # [SECTION: IMPORTS] from typing import List, Optional from sqlalchemy.orm import Session, selectinload from ...models.auth import Permission, Role, User, ADGroupMapping from ...models.profile import UserDashboardPreference from ..logger import belief_scope, logger # [/SECTION] # [DEF:AuthRepository:Class] # @PURPOSE: Provides low-level CRUD operations for identity and authorization records. # @PRE: Database session is bound. # @POST: Entity instances returned safely. # @SIDE_EFFECT: Performs database reads. # @RELATION: DEPENDS_ON -> [Session] class AuthRepository: # @PURPOSE: Initialize repository with database session. def __init__(self, db: Session): self.db = db # [DEF:get_user_by_id:Function] # @PURPOSE: Retrieve user by UUID. # @PRE: user_id is a valid UUID string. # @POST: Returns User object if found, else None. # @RELATION: DEPENDS_ON -> [User] def get_user_by_id(self, user_id: str) -> Optional[User]: with belief_scope("AuthRepository.get_user_by_id"): logger.reason(f"Fetching user by id: {user_id}") result = self.db.query(User).filter(User.id == user_id).first() logger.reflect(f"User found: {result is not None}") return result # [/DEF:get_user_by_id:Function] # [DEF:get_user_by_username:Function] # @PURPOSE: Retrieve user by username. # @PRE: username is a non-empty string. # @POST: Returns User object if found, else None. # @RELATION: DEPENDS_ON -> [User] def get_user_by_username(self, username: str) -> Optional[User]: with belief_scope("AuthRepository.get_user_by_username"): logger.reason(f"Fetching user by username: {username}") result = self.db.query(User).filter(User.username == username).first() logger.reflect(f"User found: {result is not None}") return result # [/DEF:get_user_by_username:Function] # [DEF:get_role_by_id:Function] # @PURPOSE: Retrieve role by UUID with permissions preloaded. # @RELATION: DEPENDS_ON -> [Role] # @RELATION: DEPENDS_ON -> [Permission] def get_role_by_id(self, role_id: str) -> Optional[Role]: with belief_scope("AuthRepository.get_role_by_id"): return ( self.db.query(Role) .options(selectinload(Role.permissions)) .filter(Role.id == role_id) .first() ) # [/DEF:get_role_by_id:Function] # [DEF:get_role_by_name:Function] # @PURPOSE: Retrieve role by unique name. # @RELATION: DEPENDS_ON -> [Role] def get_role_by_name(self, name: str) -> Optional[Role]: with belief_scope("AuthRepository.get_role_by_name"): return self.db.query(Role).filter(Role.name == name).first() # [/DEF:get_role_by_name:Function] # [DEF:get_permission_by_id:Function] # @PURPOSE: Retrieve permission by UUID. # @RELATION: DEPENDS_ON -> [Permission] def get_permission_by_id(self, permission_id: str) -> Optional[Permission]: with belief_scope("AuthRepository.get_permission_by_id"): return ( self.db.query(Permission).filter(Permission.id == permission_id).first() ) # [/DEF:get_permission_by_id:Function] # [DEF:get_permission_by_resource_action:Function] # @PURPOSE: Retrieve permission by resource and action tuple. # @RELATION: DEPENDS_ON -> [Permission] def get_permission_by_resource_action( self, resource: str, action: str ) -> Optional[Permission]: with belief_scope("AuthRepository.get_permission_by_resource_action"): return ( self.db.query(Permission) .filter(Permission.resource == resource, Permission.action == action) .first() ) # [/DEF:get_permission_by_resource_action:Function] # [DEF:list_permissions:Function] # @PURPOSE: List all system permissions. # @RELATION: DEPENDS_ON -> [Permission] def list_permissions(self) -> List[Permission]: with belief_scope("AuthRepository.list_permissions"): return self.db.query(Permission).all() # [/DEF:list_permissions:Function] # [DEF:get_user_dashboard_preference:Function] # @PURPOSE: Retrieve dashboard filters/preferences for a user. # @RELATION: DEPENDS_ON -> [UserDashboardPreference] def get_user_dashboard_preference( self, user_id: str ) -> Optional[UserDashboardPreference]: with belief_scope("AuthRepository.get_user_dashboard_preference"): return ( self.db.query(UserDashboardPreference) .filter(UserDashboardPreference.user_id == user_id) .first() ) # [/DEF:get_user_dashboard_preference:Function] # [DEF:get_roles_by_ad_groups:Function] # @PURPOSE: Retrieve roles that match a list of AD group names. # @PRE: groups is a list of strings representing AD group identifiers. # @POST: Returns a list of Role objects mapped to the provided AD groups. # @RELATION: DEPENDS_ON -> [Role] # @RELATION: DEPENDS_ON -> [ADGroupMapping] def get_roles_by_ad_groups(self, groups: List[str]) -> List[Role]: with belief_scope("AuthRepository.get_roles_by_ad_groups"): logger.reason(f"Fetching roles for AD groups: {groups}") if not groups: return [] return ( self.db.query(Role) .join(ADGroupMapping) .filter(ADGroupMapping.ad_group.in_(groups)) .all() ) # [/DEF:get_roles_by_ad_groups:Function] # [/DEF:AuthRepository:Class] # [/DEF:AuthRepositoryModule:Module]