Передаем на тест

backend/src/core/auth/repository.py

@@ -0,0 +1,123 @@
+# [DEF:backend.src.core.auth.repository:Module]
+#
+# @SEMANTICS: auth, repository, database, user, role
+# @PURPOSE:   Data access layer for authentication-related entities.
+# @LAYER:     Core
+# @RELATION:  DEPENDS_ON -> sqlalchemy
+# @RELATION:  USES -> backend.src.models.auth
+#
+# @INVARIANT: All database operations must be performed within a session.
+
+# [SECTION: IMPORTS]
+from typing import Optional, List
+from sqlalchemy.orm import Session
+from ...models.auth import User, Role, Permission, ADGroupMapping
+from ..logger import belief_scope
+# [/SECTION]
+
+# [DEF:AuthRepository:Class]
+# @PURPOSE: Encapsulates database operations for authentication.
+class AuthRepository:
+    # [DEF:__init__:Function]
+    # @PURPOSE: Initializes the repository with a database session.
+    # @PARAM:   db (Session) - SQLAlchemy session.
+    def __init__(self, db: Session):
+        self.db = db
+    # [/DEF:__init__:Function]
+
+    # [DEF:get_user_by_username:Function]
+    # @PURPOSE: Retrieves a user by their username.
+    # @PRE:     username is a string.
+    # @POST:    Returns User object if found, else None.
+    # @PARAM:   username (str) - The username to search for.
+    # @RETURN:  Optional[User] - The found user or None.
+    def get_user_by_username(self, username: str) -> Optional[User]:
+        with belief_scope("AuthRepository.get_user_by_username"):
+            return self.db.query(User).filter(User.username == username).first()
+    # [/DEF:get_user_by_username:Function]
+
+    # [DEF:get_user_by_id:Function]
+    # @PURPOSE: Retrieves a user by their unique ID.
+    # @PRE:     user_id is a valid UUID string.
+    # @POST:    Returns User object if found, else None.
+    # @PARAM:   user_id (str) - The user's unique identifier.
+    # @RETURN:  Optional[User] - The found user or None.
+    def get_user_by_id(self, user_id: str) -> Optional[User]:
+        with belief_scope("AuthRepository.get_user_by_id"):
+            return self.db.query(User).filter(User.id == user_id).first()
+    # [/DEF:get_user_by_id:Function]
+
+    # [DEF:get_role_by_name:Function]
+    # @PURPOSE: Retrieves a role by its name.
+    # @PRE:     name is a string.
+    # @POST:    Returns Role object if found, else None.
+    # @PARAM:   name (str) - The role name to search for.
+    # @RETURN:  Optional[Role] - The found role or None.
+    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:update_last_login:Function]
+    # @PURPOSE: Updates the last_login timestamp for a user.
+    # @PRE:     user object is a valid User instance.
+    # @POST:    User's last_login is updated in the database.
+    # @SIDE_EFFECT: Commits the transaction.
+    # @PARAM:   user (User) - The user to update.
+    def update_last_login(self, user: User):
+        with belief_scope("AuthRepository.update_last_login"):
+            from datetime import datetime
+            user.last_login = datetime.utcnow()
+            self.db.add(user)
+            self.db.commit()
+    # [/DEF:update_last_login:Function]
+
+    # [DEF:get_role_by_id:Function]
+    # @PURPOSE: Retrieves a role by its unique ID.
+    # @PRE:     role_id is a string.
+    # @POST:    Returns Role object if found, else None.
+    # @PARAM:   role_id (str) - The role's unique identifier.
+    # @RETURN:  Optional[Role] - The found role or None.
+    def get_role_by_id(self, role_id: str) -> Optional[Role]:
+        with belief_scope("AuthRepository.get_role_by_id"):
+            return self.db.query(Role).filter(Role.id == role_id).first()
+    # [/DEF:get_role_by_id:Function]
+
+    # [DEF:get_permission_by_id:Function]
+    # @PURPOSE: Retrieves a permission by its unique ID.
+    # @PRE:     perm_id is a string.
+    # @POST:    Returns Permission object if found, else None.
+    # @PARAM:   perm_id (str) - The permission's unique identifier.
+    # @RETURN:  Optional[Permission] - The found permission or None.
+    def get_permission_by_id(self, perm_id: str) -> Optional[Permission]:
+        with belief_scope("AuthRepository.get_permission_by_id"):
+            return self.db.query(Permission).filter(Permission.id == perm_id).first()
+    # [/DEF:get_permission_by_id:Function]
+
+    # [DEF:get_permission_by_resource_action:Function]
+    # @PURPOSE: Retrieves a permission by resource and action.
+    # @PRE:     resource and action are strings.
+    # @POST:    Returns Permission object if found, else None.
+    # @PARAM:   resource (str) - The resource name.
+    # @PARAM:   action (str) - The action name.
+    # @RETURN:  Optional[Permission] - The found permission or None.
+    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: Lists all available permissions.
+    # @POST:    Returns a list of all Permission objects.
+    # @RETURN:  List[Permission] - List of permissions.
+    def list_permissions(self) -> List[Permission]:
+        with belief_scope("AuthRepository.list_permissions"):
+            return self.db.query(Permission).all()
+    # [/DEF:list_permissions:Function]
+
+# [/DEF:AuthRepository:Class]
+
+# [/DEF:backend.src.core.auth.repository:Module]