110 lines
4.8 KiB
Python
110 lines
4.8 KiB
Python
from sqlalchemy import select
|
|
from db.handlers.toolkit import ToolkitHandler
|
|
from utils import logger
|
|
from db import CRUD
|
|
from db.schemas.categories import Category
|
|
from db.handlers.records import ServiceRecordsHandler
|
|
|
|
|
|
class CategoryHandler:
|
|
async def add(newCategoryData: dict, user_id: int = None):
|
|
title = newCategoryData.get("title", None)
|
|
if not title:
|
|
logger.error("Не указано название категории")
|
|
return {}
|
|
query = select(Category).where(Category.title == title)
|
|
category = await CRUD.read(query)
|
|
if category:
|
|
logger.error("Категория с таким названием уже существует")
|
|
return {}
|
|
try:
|
|
newCategory = await Category(**newCategoryData).save()
|
|
except Exception as e:
|
|
logger.error(f"Ошибка сохранения категории: {str(e)}")
|
|
return {}
|
|
if not newCategory:
|
|
logger.error("Категория не сохранена")
|
|
return {}
|
|
await ServiceRecordsHandler.add(
|
|
user_id, {"Добавлена категория": newCategory.toDict()}
|
|
)
|
|
logger.info(
|
|
f"Категория {newCategory.title} успешно добавлена, id: {newCategory.id}"
|
|
)
|
|
return newCategory.toDict()
|
|
|
|
async def edit(categoryData: dict, userId: int):
|
|
categoryId = categoryData.pop("id", None)
|
|
if not categoryId:
|
|
logger.error("Не указан id категории")
|
|
return {}
|
|
query = select(Category).where(Category.id == categoryId)
|
|
category = await CRUD.read(query)
|
|
if not category:
|
|
logger.error("Категория не найдена")
|
|
return {}
|
|
try:
|
|
logger.info(
|
|
f"Обновление категории {category.title} -> {categoryData.get('title')}"
|
|
)
|
|
editedCategory = await category.edit(**categoryData)
|
|
except Exception as e:
|
|
logger.error(f"Ошибка обновления категории: {str(e)}")
|
|
return {}
|
|
if not editedCategory:
|
|
logger.error("Категория не обновлена")
|
|
return {}
|
|
await ServiceRecordsHandler.add(
|
|
userId, {f"Обновлена категория {category.title}": editedCategory.toDict()}
|
|
)
|
|
logger.info(f"Категория {editedCategory.title} успешно обновлена")
|
|
return editedCategory.toDict()
|
|
|
|
async def getAll() -> list[dict]:
|
|
query = select(Category)
|
|
categories = await CRUD.read(query, True)
|
|
return [category.toDict() for category in categories] if categories else []
|
|
|
|
async def getSeveral(categoryIds: list[int]) -> list[dict]:
|
|
query = select(Category).where(Category.id.in_(categoryIds))
|
|
categories = await CRUD.read(query, True)
|
|
return [category.toDict() for category in categories] if categories else []
|
|
|
|
async def delete(categoryId: int, user_id: int = None):
|
|
categoryInUse = await ToolkitHandler.checkCatogoryUse(categoryId)
|
|
if categoryInUse:
|
|
logger.error("Категория используется в инструментах")
|
|
return True
|
|
query = select(Category).where(Category.id == categoryId)
|
|
category = await CRUD.read(query)
|
|
if not category:
|
|
logger.error("Категория не найдена")
|
|
return False
|
|
try:
|
|
categoryTitle = category.title
|
|
result = await CRUD.delete(category)
|
|
except Exception as e:
|
|
logger.error(f"Ошибка удаления категории: {str(e)}")
|
|
return False
|
|
await ServiceRecordsHandler.add(
|
|
user_id, {"Удалена категория": f"Название: {categoryTitle}"}
|
|
)
|
|
logger.info(
|
|
f"Категория {categoryTitle} {'успешно удалена' if result else 'не удалена'}"
|
|
)
|
|
return result
|
|
|
|
async def initialize():
|
|
baseCategories = [
|
|
{"title": "Фрезеровка", "description": "Инструмент для фрезерного цеха"},
|
|
{"title": "Токарка", "description": "Инструмент для токарного цеха"},
|
|
{"title": "Слесарка", "description": "Инструмент для слесарного цеха"},
|
|
]
|
|
|
|
logger.info("Создание базовых категорий")
|
|
for categoryData in baseCategories:
|
|
await CategoryHandler.add(categoryData)
|
|
|
|
logger.info("Категории успешно созданы")
|
|
return
|