101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
from typing import Optional
|
||
|
||
from fastapi import HTTPException
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from starlette import status
|
||
|
||
from app.application.lenses_repository import LensesRepository
|
||
from app.application.set_content_repository import SetContentRepository
|
||
from app.application.sets_repository import SetsRepository
|
||
from app.domain.entities.lens import LensEntity
|
||
from app.domain.entities.set import SetEntity
|
||
from app.domain.models import Set, Lens
|
||
from app.infrastructure.lenses_service import LensesService
|
||
|
||
|
||
class SetsService:
|
||
def __init__(self, db: AsyncSession):
|
||
self.sets_repository = SetsRepository(db)
|
||
self.set_content_repository = SetContentRepository(db)
|
||
self.lenses_repository = LensesRepository(db)
|
||
|
||
async def get_all_sets(self) -> list[SetEntity]:
|
||
sets = await self.sets_repository.get_all()
|
||
|
||
return [
|
||
self.model_to_entity(_set)
|
||
for _set in sets
|
||
]
|
||
|
||
async def create_set(self, _set: SetEntity) -> SetEntity:
|
||
set_model = self.entity_to_model(_set)
|
||
|
||
await self.sets_repository.create(set_model)
|
||
|
||
return self.model_to_entity(set_model)
|
||
|
||
async def update_set(self, set_id: int, _set: SetEntity) -> Optional[SetEntity]:
|
||
set_model = await self.sets_repository.get_by_id(set_id)
|
||
|
||
if not set_model:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Набор линз не найден")
|
||
|
||
set_model.title = _set.title
|
||
|
||
await self.sets_repository.update(set_model)
|
||
|
||
return self.model_to_entity(set_model)
|
||
|
||
async def delete_set(self, set_id: int) -> Optional[SetEntity]:
|
||
_set = await self.sets_repository.get_by_id(set_id)
|
||
|
||
if not _set:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Набор линз не найден")
|
||
|
||
set_content = await self.set_content_repository.get_by_set_id(set_id)
|
||
await self.set_content_repository.delete_list_sets(
|
||
list(set_content)
|
||
)
|
||
|
||
result = await self.sets_repository.delete(_set)
|
||
|
||
return self.model_to_entity(result)
|
||
|
||
async def append_set_content_to_lenses(self, set_id: int) -> Optional[list[LensEntity]]:
|
||
_set = await self.sets_repository.get_by_id(set_id)
|
||
|
||
if not _set:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Набор линз не найден")
|
||
|
||
set_content = await self.set_content_repository.get_by_set_id(set_id)
|
||
|
||
lenses = []
|
||
for content in set_content:
|
||
lens = LensesService.entity_to_model(content)
|
||
|
||
await self.lenses_repository.create(lens)
|
||
|
||
lenses.append(
|
||
LensesService.model_to_entity(lens)
|
||
)
|
||
|
||
return lenses
|
||
|
||
@staticmethod
|
||
def model_to_entity(_set: Set) -> SetEntity:
|
||
return SetEntity(
|
||
id=_set.id,
|
||
title=_set.title,
|
||
)
|
||
|
||
@staticmethod
|
||
def entity_to_model(_set: SetEntity) -> Set:
|
||
set_model = Set(
|
||
title=_set.title,
|
||
)
|
||
|
||
if _set.id is not None:
|
||
set_model.id = _set.id
|
||
|
||
return set_model
|