83372f7aef
Конфликт в api/tests/integration/test_form_phases_api_smoke.py: оставлены оба набора тестов (тесты фильтра по роли и 404 из этого PR и тест aware-дат из test). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.db.models.form3_phase import Form3Phase
|
|
from src.db.models.form_type import FormType, FormTypeEnum
|
|
from src.repository.form3_phase_repository import Form3PhaseRepository
|
|
from src.core.timezones import to_org_local_naive
|
|
from src.core.errors import AccessDeniedException, ValidationsError
|
|
from src.db.models.role import UserRoleEnum
|
|
from src.db.models.app_user import AppUser
|
|
from src.domain.schemas import Form3PhaseCreate, Form3PhaseUpdate
|
|
from src.services.form_phase_service import role_code_for_user
|
|
|
|
|
|
class Form3PhaseService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
self.f3p_repo = Form3PhaseRepository(db)
|
|
|
|
async def get_list(
|
|
self,
|
|
report_id: int | list[int],
|
|
user: AppUser,
|
|
phase_code: str | None = None,
|
|
) -> list[Form3Phase]:
|
|
return await self.f3p_repo.get_list(
|
|
report_id=report_id,
|
|
phase_code=phase_code,
|
|
role=role_code_for_user(user),
|
|
)
|
|
|
|
async def get(
|
|
self, report_id: int, phase_code: str, user: AppUser
|
|
) -> Form3Phase | None:
|
|
return await self.f3p_repo.get(
|
|
report_id=report_id,
|
|
phase_code=phase_code,
|
|
)
|
|
|
|
def _validate_form3_phase(
|
|
self,
|
|
form_phase: Form3PhaseCreate | Form3PhaseUpdate,
|
|
) -> None:
|
|
errors = {}
|
|
if form_phase.column_keys:
|
|
form_type = FormType(code=FormTypeEnum.FORM_3)
|
|
for column_key in form_phase.column_keys:
|
|
section = column_key.split(".")
|
|
if section[0] not in 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,
|
|
report_id: int,
|
|
body: Form3PhaseCreate,
|
|
user: AppUser,
|
|
utc_offset: str | None = None,
|
|
) -> Form3Phase:
|
|
if user.role_id != UserRoleEnum.ADMIN:
|
|
raise AccessDeniedException
|
|
|
|
self._validate_form3_phase(
|
|
form_phase=body,
|
|
)
|
|
opens_at = to_org_local_naive(body.opens_at, utc_offset)
|
|
closes_at = to_org_local_naive(body.closes_at, utc_offset)
|
|
return await self.f3p_repo.create(
|
|
report_id=report_id,
|
|
phase_code=body.phase_code,
|
|
role=body.role,
|
|
column_keys=body.column_keys,
|
|
opens_at=opens_at,
|
|
closes_at=closes_at,
|
|
)
|
|
|
|
async def update(
|
|
self,
|
|
report_id: int,
|
|
phase_code: str,
|
|
body: Form3PhaseUpdate,
|
|
user: AppUser,
|
|
utc_offset: str | None = None,
|
|
) -> Form3Phase:
|
|
if user.role_id != UserRoleEnum.ADMIN:
|
|
raise AccessDeniedException
|
|
|
|
self._validate_form3_phase(
|
|
form_phase=body,
|
|
)
|
|
|
|
data = body.model_dump(exclude_unset=True)
|
|
for key in ("opens_at", "closes_at"):
|
|
if data.get(key) is not None:
|
|
data[key] = to_org_local_naive(data[key], utc_offset)
|
|
if not data:
|
|
return await self.f3p_repo.get(
|
|
report_id=report_id,
|
|
phase_code=phase_code,
|
|
)
|
|
return await self.f3p_repo.update(
|
|
report_id=report_id,
|
|
phase_code=phase_code,
|
|
data=data,
|
|
)
|
|
|
|
async def delete(
|
|
self, report_id: int, phase_code: str, user: AppUser,
|
|
) -> bool:
|
|
return await self.f3p_repo.delete(
|
|
report_id=report_id,
|
|
phase_code=phase_code,
|
|
)
|