toolbox/db/handlers/access.py

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


class AccessLevelHandler:
    async def add(newData, userId: int = None):
        title = newData.get("title", None)
        if not title:
            logger.error("Не указано название уровня доступа")
            return {"error": "Не указано название уровня доступа"}
        exists = await CRUD.read(select(AccessLevel).where(AccessLevel.title == title))
        if exists:
            logger.error("Уровень доступа с таким названием уже существует")
            return {"error": "Уровень доступа с таким названием уже существует"}
        try:
            logger.info(f"Создание уровня доступа {title}")
            accessData = await AccessLevel(**newData).save()
            await ServiceRecordsHandler.add(
                userId, {"Добавлен уровень доступа": 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(levelData, userId: int = None):
        query = select(AccessLevel).where(AccessLevel.id == levelData.pop("id"))
        accessData = await CRUD.read(query)
        if not accessData:
            logger.error("Уровень доступа не найден")
            return {"error": "Уровень доступа не найден"}
        try:
            editedAccessData = await accessData.edit(**levelData)
            await ServiceRecordsHandler.add(
                userId, {"Обновлен уровень доступа": editedAccessData.toDict()}
            )
        except Exception as e:
            logger.error(f"Ошибка обновления уровня доступа: {str(e)}")
            return {"error": "Ошибка обновления уровня доступа"}
        logger.info(
            f"Уровень доступа {editedAccessData.title} успешно обновлен, изменены данные: {levelData.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():
        baseAcessLevels = {
            "admin": {
                "title": "Администратор",
                "description": "Администратор. Полный доступ",
                "refund_request_confirm": True,
                "debit_request_confirm": True,
                "tools_creation": True,
                "tools_registration": True,
                "tools_edit": True,
                "tools_delete": True,
                "users_creation": True,
                "users_edit": True,
                "users_disabling": True,
                "users_view": True,
                "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,
            },
        }
        logger.info("Инициализация уровней доступа")

        for accessLevel in baseAcessLevels.values():
            await AccessLevelHandler.add(accessLevel)

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