services/authorize/database_utils.py - dev - Arslan16/FleetHub

services/authorize/database_utils.py
141 строка5 KB
Arslan16
Добавил в python сервисах почти везде docstrings. В lifespan добавил в except ошибки OSError и CannotConnectNowError
03 авг 2026, 23:31
03 авг 2026, 23:3178eade6
"""
Модуль управления данными пользователей сервиса Authorize.

Содержит менеджер для работы с ORM-моделью UsersTable:
- проверка существования пользователей;
- создание учетных записей;
- проверка корректности пароля;
- получение пользователей по логину.

Использует асинхронную работу с PostgreSQL через SQLAlchemy AsyncSession.
"""
import hashlib

from sqlalchemy import insert, select, exists
from sqlalchemy.sql import Insert, Select
from sqlalchemy.ext.asyncio import AsyncSession

from schemas import UserContext
from models import UsersTable
from logger import logger


class UsersManager:
    def __init__(self, session: AsyncSession = None):
        self.session = session

    async def is_user_exists(
        self,
        login: str
    ):
        """
        Проверяет наличие пользователя по логину

        Args:
            login (str): Логин пользователя

        Returns:
            result (bool): Результат проверки. В случае ошибки выполнения возвращает `True` для предотвращения\n
            регистрации существующего пользователя
        """
        try:
            stmt: Select = select(
                exists(
                    UsersTable.login
                )
            ).where(
                UsersTable.login == login
            )
            return await self.session.scalar(stmt)
        except Exception as exc:
            logger.exception(exc, {"login": login})
            return True # Чтоб не было регистраций, пока есть ошибки

    async def create_user(
        self,
        login: str,
        password: str,
        role: str = UserContext.UserRoles.client.value
    ) -> int | None:
        """
        Создает нового пользователя с введенными данными

        Args:
            login (str): Логин нового пользователя
            password (str): Исходный пароль пользователя (Хэшируется перед записью в бд)

        Returns:
            result (int): Идентификатор созданного пользователя или None в случае ошибки
        """
        try:
            stmt: Insert = insert(
                UsersTable
            ).values(
                login=login,
                password=hashlib.sha256(password.encode("utf8")).hexdigest(),
                role=role
            ).returning(
                UsersTable.id
            )
            user_id = await self.session.scalar(stmt)
            return user_id
        except Exception as exc:
            logger.exception(exc, {"login": login, "role": role})

    async def is_password_correct(
        self,
        login: str,
        password: str
    ) -> bool:
        """
        Проверяет корректность пароля пользователя

        Args:
            login (str): Логин пользователя
            password (str): Исходный пароль пользователя (Хэшируется перед проверкой)

        Returns:
            result (bool): результат проверки. `False` в случае ошибки
        """
        try:
            stmt: Select = select(
                exists(
                    UsersTable.id
                )
            ).where(
                UsersTable.login == login,
                UsersTable.password == hashlib.sha256(password.encode()).hexdigest()
            )
            return await self.session.scalar(stmt)
        except Exception as exc:
            logger.exception(exc, {"login": login})
            return False

    async def get_user_by_login(
        self,
        login: str
    ) -> int | None:
        """
        Извлекает id пользователя по его логину. (Без проверок пароля)

        Args:
            login (str): Логин пользователя

        Returns:
            result (int | None): ИД пользователя или None. В случае ошибки - None
        """
        try:
            stmt: Select = select(
                UsersTable
            ).where(
                UsersTable.login == login
            )
            return await self.session.scalar(stmt)
        except Exception as exc:
            logger.exception(exc, {"login": login})
            return None


__all__: list[str] = [
    "UsersManager"
]