toolbox/db/handlers/access.py

from sqlalchemy import select
from utils import logger
from db import CRUD
from db.schemas import AccessLevel
from db.handlers import ServiceRecordsHandler


class AccessLevelHandler:
    async def add(**kwargs):
        title = kwargs.get("title", None)
        if not title:
            logger.error("Не указано название уровня доступа")
            return {}
        exists = await CRUD.read(select(AccessLevel).where(AccessLevel.title == title))
        if exists:
            logger.error("Уровень доступа с таким названием уже существует")
            return {}
        try:
            logger.info(f"Создание уровня доступа {title}")
            user_id = kwargs.pop("user_id", None)
            accessData = await AccessLevel(**kwargs).save()
            await ServiceRecordsHandler.add(
                user_id, {"Добавлен уровень доступа": accessData.toDict()}
            )
        except Exception as e:
            logger.error(f"Ошибка создания уровня доступа: {str(e)}")
            return {}
        logger.info(f"Уровень доступа {accessData.title} успешно создан")
        return accessData.toDict()

    async def get(accessId: int) -> dict:
        query = select(AccessLevel).where(AccessLevel.id == accessId)
        accessData = await CRUD.read(query)
        if not accessData:
            logger.error("Уровень доступа не найден")
            return {}
        return accessData.toDict()

    async def edit(accessId: int, **kwargs):
        query = select(AccessLevel).where(AccessLevel.id == accessId)
        accessData = await CRUD.read(query)
        if not accessData:
            logger.error("Уровень доступа не найден")
            return {}
        try:
            user_id = kwargs.pop("user_id", None)
            editedAccessData = await accessData.edit(**kwargs)
            await ServiceRecordsHandler.add(
                user_id, {"Обновлен уровень доступа": editedAccessData.toDict()}
            )
        except Exception as e:
            logger.error(f"Ошибка обновления уровня доступа: {str(e)}")
            return {}
        logger.info(
            f"Уровень доступа {editedAccessData.title} успешно обновлен, изменены данные: {kwargs.keys()}"
        )
        return editedAccessData.toDict()

    async def getAll() -> list:
        query = select(AccessLevel)
        accessLevels = await CRUD.read(query, True)
        return (
            [accessLevel.toDict() for accessLevel in accessLevels]
            if accessLevels
            else []
        )

    async def delete(accessId: int, user_id: int = None):
        query = select(AccessLevel).where(AccessLevel.id == accessId)
        accessData = await CRUD.read(query)
        if not accessData:
            logger.error("Уровень доступа не найден")
            return False
        try:
            title = accessData.title
            result = await CRUD.delete(accessData)
            await ServiceRecordsHandler.add(
                user_id, {"Удален уровень доступа": f"Название: {title}"}
            )
        except Exception as e:
            logger.error(f"Ошибка удаления уровня доступа: {str(e)}")
            return False
        logger.info(
            f"Уровень доступа {accessData.title} {'успешно удален' if result else 'не удален'}"
        )
        return result

    async def initialize(self):
        baseAcessLevels = {
            "admin": {
                "title": "Администратор",
                "description": "Администратор. Полный доступ",
                "receiving_edit": True,
                "refund_request_edit": True,
                "refund_request_confirm": True,
                "debit_request_edit": True,
                "debit_request_confirm": True,
                "tools_creation": True,
                "tools_registration": True,
                "tools_registration_edit": True,
                "tools_edit": True,
                "tools_delete": True,
                "users_creation": True,
                "users_edit": True,
                "users_disabling": True,
                "users_view": True,
                "available_own_toolbox": False,
                "view_all_toolboxes": True,
                "view_requests": True,
                "view_services": True,
                "access_level_view": True,
                "access_level_edit": True,
                "manage_toolboxes": True,
            },
            "manager": {
                "title": "Менеджер",
                "description": "Менеджер. Доступ к просмотру и редактированию рабочей информации",
                "refund_request_confirm": True,
                "debit_request_confirm": True,
                "tools_creation": True,
                "tools_registration": True,
                "tools_edit": True,
                "users_disabling": True,
                "users_view": True,
                "view_all_toolboxes": True,
                "view_requests": True,
                "view_services": True,
                "access_level_view": True,
                "manage_toolboxes": True,
            },
            "storekeeper": {
                "title": "Кладовщик",
                "description": "Кладовщик. Доступ к управлению складом",
                "refund_request_confirm": True,
                "debit_request_confirm": True,
                "tools_creation": True,
                "tools_registration": True,
                "tools_edit": True,
                "users_view": True,
                "view_requests": True,
                "view_all_toolboxes": True,
            },
            "employee": {
                "title": "Сотрудник",
                "description": "Сотрудник. Управление собственной рабочей информацией",
                "available_own_toolbox": True,
            },
        }

        for accessLevel in baseAcessLevels.values():
            await self.add(**accessLevel)

        logger.info("Уровни доступа успешно инициализированы")
        return