137 lines
6.2 KiB
Python
137 lines
6.2 KiB
Python
# [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 ->[sqlalchemy.orm.Session]
|
|
# @RELATION: DEPENDS_ON ->[User:Class]
|
|
# @RELATION: DEPENDS_ON ->[Role:Class]
|
|
# @RELATION: DEPENDS_ON ->[Permission:Class]
|
|
# @RELATION: DEPENDS_ON ->[UserDashboardPreference:Class]
|
|
# @RELATION: DEPENDS_ON ->[belief_scope:Function]
|
|
# @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 -> sqlalchemy.orm.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]
|