140 lines
4.8 KiB
Python
140 lines
4.8 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.core.errors import AccessDeniedException, ValidationsError
|
||
from src.db.models.budget_form import BudgetForm
|
||
from src.db.models.role import UserRoleEnum
|
||
from src.repository.budget_form_repository import BudgetFormRepository
|
||
from src.db.models.app_user import AppUser
|
||
from src.db.models.form_phase import FormPhase
|
||
from src.domain.schemas import FormPhaseCreate, FormPhaseUpdate
|
||
from src.repository.form_phase_repository import FormPhaseRepository
|
||
|
||
|
||
class FormPhaseService:
|
||
def __init__(self, db: AsyncSession):
|
||
self.db = db
|
||
self.fp_repo = FormPhaseRepository(db)
|
||
self.bf_repo = BudgetFormRepository(db)
|
||
|
||
async def get_list(
|
||
self,
|
||
budget_form_id: int,
|
||
user: AppUser,
|
||
sheet: str | None = None,
|
||
phase_code: str | None = None,
|
||
) -> list[FormPhase]:
|
||
return await self.fp_repo.get_list(
|
||
budget_form_id=budget_form_id,
|
||
sheet=sheet,
|
||
phase_code=phase_code,
|
||
)
|
||
|
||
async def get(
|
||
self, budget_form_id: int, sheet: str, phase_code: str, user: AppUser
|
||
) -> FormPhase | None:
|
||
return await self.fp_repo.get(
|
||
budget_form_id=budget_form_id,
|
||
sheet=sheet,
|
||
phase_code=phase_code,
|
||
)
|
||
|
||
def _validate_form_phase(
|
||
self,
|
||
form_phase: FormPhaseCreate | FormPhaseUpdate,
|
||
budget_form: BudgetForm,
|
||
sheet: str | None = None,
|
||
) -> None:
|
||
errors = {}
|
||
if budget_form.form_type.sheet_list:
|
||
sheet = sheet or form_phase.sheet
|
||
if sheet not in budget_form.form_type.sheet_list:
|
||
errors["sheet"] = [f"У формы нет листа {sheet}"]
|
||
if form_phase.column_keys and budget_form.form_type.section_list:
|
||
for column_key in form_phase.column_keys:
|
||
section = column_key.split(".")
|
||
if section[0] not in budget_form.form_type.section_list:
|
||
if "column_keys" in errors:
|
||
errors["column_keys"].append(f"Колонка {column_key} некорректна")
|
||
else:
|
||
errors["column_keys"] = [f"Колонка {column_key} некорректна"]
|
||
if errors:
|
||
raise ValidationsError(errors=errors)
|
||
|
||
|
||
async def create(
|
||
self,
|
||
body: FormPhaseCreate,
|
||
user: AppUser,
|
||
budget_form_id: int | None = None,
|
||
budget_form: BudgetForm | None = None,
|
||
) -> FormPhase:
|
||
assert budget_form or budget_form_id
|
||
if user.role_id != UserRoleEnum.ADMIN:
|
||
raise AccessDeniedException
|
||
|
||
if not budget_form or budget_form.form_type is None:
|
||
budget_form = await self.bf_repo.get(
|
||
budget_form_id=budget_form_id or budget_form.id,
|
||
load_form_type=True,
|
||
)
|
||
self._validate_form_phase(
|
||
form_phase=body,
|
||
budget_form=budget_form,
|
||
)
|
||
return await self.fp_repo.create(
|
||
budget_form_id=budget_form_id or budget_form.id,
|
||
sheet=body.sheet,
|
||
phase_code=body.phase_code,
|
||
role=body.role,
|
||
column_keys=body.column_keys,
|
||
opens_at=body.opens_at,
|
||
closes_at=body.closes_at,
|
||
)
|
||
|
||
async def update(
|
||
self,
|
||
sheet: str,
|
||
phase_code: str,
|
||
body: FormPhaseUpdate,
|
||
user: AppUser,
|
||
budget_form_id: int | None = None,
|
||
budget_form: BudgetForm | None = None,
|
||
) -> FormPhase:
|
||
assert budget_form or budget_form_id
|
||
if user.role_id != UserRoleEnum.ADMIN:
|
||
raise AccessDeniedException
|
||
|
||
if not budget_form or budget_form.form_type is None:
|
||
budget_form = await self.bf_repo.get(
|
||
budget_form_id=budget_form_id or budget_form.id,
|
||
load_form_type=True,
|
||
)
|
||
self._validate_form_phase(
|
||
form_phase=body,
|
||
budget_form=budget_form,
|
||
sheet=sheet,
|
||
)
|
||
|
||
data = body.model_dump(exclude_unset=True)
|
||
if not data:
|
||
return await self.fp_repo.get(
|
||
budget_form_id=budget_form_id or budget_form.id,
|
||
sheet=sheet,
|
||
phase_code=phase_code,
|
||
)
|
||
return await self.fp_repo.update(
|
||
budget_form_id=budget_form_id or budget_form.id,
|
||
sheet=sheet,
|
||
phase_code=phase_code,
|
||
data=data,
|
||
)
|
||
|
||
async def delete(
|
||
self, budget_form_id: int, sheet: str, phase_code: str, user: AppUser
|
||
) -> bool:
|
||
return await self.fp_repo.delete(
|
||
budget_form_id=budget_form_id,
|
||
sheet=sheet,
|
||
phase_code=phase_code,
|
||
)
|