159 lines
6.0 KiB
Python
159 lines
6.0 KiB
Python
from typing import Optional
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.application.contest_statuses_repository import ContestStatusesRepository
|
|
from app.application.contests_repository import ContestsRepository
|
|
from app.application.projects_repository import ProjectsRepository
|
|
from app.application.users_repository import UsersRepository
|
|
from app.domain.entities.contest import ContestEntity
|
|
from app.domain.models import Contest, User
|
|
|
|
|
|
class ContestsService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.contests_repository = ContestsRepository(db)
|
|
self.projects_repository = ProjectsRepository(db)
|
|
self.statuses_repository = ContestStatusesRepository(db)
|
|
self.users_repository = UsersRepository(db)
|
|
|
|
async def get_all_contests(self) -> list[ContestEntity]:
|
|
contests = await self.contests_repository.get_all()
|
|
return [
|
|
self.model_to_entity(contest)
|
|
for contest in contests
|
|
]
|
|
|
|
async def create_contest(self, contest: ContestEntity) -> Optional[ContestEntity]:
|
|
project = await self.projects_repository.get_by_id(contest.project_id)
|
|
if project is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail='The project with this ID was not found',
|
|
)
|
|
|
|
status_contest = await self.statuses_repository.get_by_id(contest.status_id)
|
|
if status_contest is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail='The status with this ID was not found',
|
|
)
|
|
|
|
contest_model = self.entity_to_model(contest)
|
|
|
|
contest_model = await self.contests_repository.create(contest_model)
|
|
|
|
return self.model_to_entity(contest_model)
|
|
|
|
async def update_contest(self, contest_id: int, contest: ContestEntity, user: User) -> Optional[
|
|
ContestEntity
|
|
]:
|
|
user = await self.users_repository.get_by_id(user.id)
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail='The user with this ID was not found',
|
|
)
|
|
elif user.profile.role.title != 'Администратор':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Permission denied",
|
|
)
|
|
|
|
contest_model = await self.contests_repository.get_by_id(contest_id)
|
|
if contest_model is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail='The contest with this ID was not found',
|
|
)
|
|
|
|
project = await self.projects_repository.get_by_id(contest.project_id)
|
|
if project is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail='The project with this ID was not found',
|
|
)
|
|
|
|
status_contest = await self.statuses_repository.get_by_id(contest.role_id)
|
|
if status_contest is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail='The status with this ID was not found',
|
|
)
|
|
|
|
contest_model.title = contest.title
|
|
contest_model.description = contest.description
|
|
contest_model.web_url = contest.web_url
|
|
contest_model.photo = contest.photo
|
|
contest_model.results = contest.results
|
|
contest_model.is_win = contest.is_win
|
|
contest_model.project_id = contest.project_id
|
|
contest_model.status_id = contest.status_id
|
|
|
|
contest_model = await self.contests_repository.update(contest_model)
|
|
|
|
return self.model_to_entity(contest_model)
|
|
|
|
async def delete(self, contest_id: int, user: User):
|
|
user = await self.users_repository.get_by_id(user.id)
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail='The user with this ID was not found',
|
|
)
|
|
|
|
contest_model = await self.contests_repository.get_by_id(contest_id)
|
|
if user.profile.role.title != 'Администратор':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail='Permission denied',
|
|
)
|
|
|
|
result = await self.contests_repository.delete(contest_model)
|
|
|
|
return self.model_to_entity(result)
|
|
|
|
async def get_by_project(self, project_id: int) -> Optional[ContestEntity]:
|
|
project = await self.projects_repository.get_by_id(project_id)
|
|
if project is None:
|
|
raise HTTPException(status_code=404, detail='Project not found')
|
|
|
|
contest_model = await self.contests_repository.get_by_id(project.contest_id)
|
|
if not contest_model:
|
|
raise HTTPException(status_code=404, detail='Contest not found')
|
|
|
|
return self.model_to_entity(contest_model)
|
|
|
|
@staticmethod
|
|
def model_to_entity(contest_model: Contest) -> ContestEntity:
|
|
return ContestEntity(
|
|
id=contest_model.id,
|
|
title=contest_model.title,
|
|
description=contest_model.description,
|
|
web_url=contest_model.web_url,
|
|
photo=contest_model.photo,
|
|
results=contest_model.results,
|
|
is_win=contest_model.is_win,
|
|
project_id=contest_model.project_id,
|
|
status_id=contest_model.status_id,
|
|
)
|
|
|
|
@staticmethod
|
|
def entity_to_model(contest_entity: ContestEntity) -> Contest:
|
|
contest_model = Contest(
|
|
title=contest_entity.title,
|
|
description=contest_entity.description,
|
|
web_url=contest_entity.web_url,
|
|
photo=contest_entity.photo,
|
|
results=contest_entity.results,
|
|
is_win=contest_entity.is_win,
|
|
project_id=contest_entity.project_id,
|
|
status_id=contest_entity.status_id
|
|
)
|
|
|
|
if contest_entity.id is not None:
|
|
contest_model.id = contest_entity.id
|
|
|
|
return contest_model
|