"""Business logic for DayAssertion — the intent-vs-derived-output restructure's single record of "what the user asserts about one calendar day" (see `app/entities/day_assertion.py`). Restructure step 3b: this service is now the ONE write path behind `POST /api/day-assertions`, superseding both the old hand-pin write on `/api/planned-days` and `/api/planned-day-exclusions` (both deleted). It owns the natural-key upsert convention AND mirrors every write into `planned_leave_day`, the same way `PlannedDayExclusionService` used to for the `'none'` case — a `DayAssertion` for a concrete `day_type` is also, immediately, a `planned_leave_day` row with that same day_type/ is_dubbeldag/wish_id (so the calendar renders it without waiting for the next `POST /api/plan/sync`), and a `day_type='none'` assertion still deletes whatever `planned_leave_day` row occupied that slot. Enforces the SAME guard `PlanningService.add_planned_day`/`update` used to (`app.services.day_guards.reject_illegal_free_day` — a free day can never be `vacation`, and may be `sjukpenning`/`lagsta` only when the workspace opted in) and the same per-day dubbeldagar-window check (`AnalyticsService.validate_dubbeldagar`), so hand-pinning a day through this endpoint is exactly as guarded as the legacy `/api/planned-days` write was — both optional constructor deps, like `PlanningService`'s own guards, so a caller that never wires them (older direct-construction call sites) keeps whatever behaviour it had. """ from datetime import date, datetime from uuid import UUID, uuid4 from app.entities.day_assertion import DAY_TYPE_NONE, DayAssertion from app.entities.planned_leave import PlannedLeaveDay from app.entities.workspace import Workspace from app.repositories.day_assertion_repository import DayAssertionRepository from app.repositories.day_override_repository import DayOverrideRepository from app.repositories.planned_leave_repository import PlannedLeaveRepository from app.repositories.unit_of_work import NoOpUnitOfWork, UnitOfWork from app.repositories.workspace_repository import WorkspaceRepository from app.services.analytics_service import AnalyticsService from app.services.day_guards import reject_illegal_free_day from app.utils.errors import NotFoundError, ValidationError _VALID_DAY_TYPES = frozenset({"sjukpenning", "lagsta", "vacation", DAY_TYPE_NONE}) class DayAssertionService: def __init__( self, repo: DayAssertionRepository, planned_leave_repo: PlannedLeaveRepository | None = None, analytics_service: AnalyticsService | None = None, workspace_repo: WorkspaceRepository | None = None, day_override_repo: DayOverrideRepository | None = None, uow: UnitOfWork | None = None, ) -> None: self._repo = repo # Only needed for the `planned_leave_day` mirroring below — optional # so a direct-construction caller that only wants pure assertion # semantics (tests) doesn't have to wire it. self._planned_leave_repo = planned_leave_repo self._analytics = analytics_service self._workspace_repo = workspace_repo self._day_override_repo = day_override_repo # Adversarial-review fix (restructure step 3b): `assert_day`/`clear` # each do an assertion write followed by a `planned_leave_day` mirror # write (create-or-replace, or a delete) — two separate commits with # no unit of work, exactly the shape `PlanningService` wraps in # `self._uow()` everywhere else. Defaults to a no-op so existing # direct-construction call sites (tests) keep today's per-write-commit # behavior; `app.dependencies.get_day_assertion_service` always wires # a real `SQLiteUnitOfWork`. self._uow = uow or NoOpUnitOfWork() async def _reject_illegal_free_day(self, workspace_id: UUID, d: date, day_type: str) -> None: if self._workspace_repo is None or self._day_override_repo is None: return overrides = { o.date: o.is_working_day for o in await self._day_override_repo.list_by_workspace(workspace_id) } workspace: Workspace | None = await self._workspace_repo.get_by_id(workspace_id) reject_illegal_free_day(workspace, overrides, d, day_type) async def assert_day( self, workspace_id: UUID, child_id: UUID, parent_personal_data_uuid: UUID, day: date, day_type: str, is_dubbeldag: bool = False, wish_id: UUID | None = None, ) -> DayAssertion: """Record (or overwrite) the user's assertion for this (workspace, child, parent, date), and mirror it into `planned_leave_day` so the calendar reflects it immediately. `day_type='none'` is the exclusion-shaped assertion — "the planner may not place leave here" — and mirrors by DELETING any `planned_leave_day` row at this key instead. Idempotent by natural key: asserting the same slot again overwrites rather than duplicating, and always OVERWRITES whatever assertion already exists there (never merges), matching `PlannedDayExclusionService.exclude()`'s precedent. """ if day_type not in _VALID_DAY_TYPES: raise ValidationError(f"invalid day_type for a DayAssertion: {day_type!r}") await self._reject_illegal_free_day(workspace_id, day, day_type) if is_dubbeldag and self._analytics is not None: # `DubbeldagError` is itself a `ValidationError` (see # `app.utils.errors`) — nothing to translate, matches # `PlanningService.add_planned_day`'s own Validation 1 exactly. await self._analytics.validate_dubbeldagar(child_id, day) now = datetime.now() async with self._uow(): created = await self._repo.upsert( DayAssertion( id=uuid4(), workspace_id=workspace_id, child_id=child_id, parent_personal_data_uuid=parent_personal_data_uuid, date=day, day_type=day_type, is_dubbeldag=is_dubbeldag, wish_id=wish_id, created_at=now, updated_at=now, ) ) await self._mirror_into_planned_leave( workspace_id, child_id, parent_personal_data_uuid, day, day_type, is_dubbeldag, wish_id, ) return created async def _mirror_into_planned_leave( self, workspace_id: UUID, child_id: UUID, parent_personal_data_uuid: UUID, day: date, day_type: str, is_dubbeldag: bool, wish_id: UUID | None, ) -> None: if self._planned_leave_repo is None: return existing = [ row for row in await self._planned_leave_repo.list_by_workspace_child( workspace_id, child_id, from_date=day, to_date=day ) if row.parent_personal_data_uuid == parent_personal_data_uuid ] if existing: await self._planned_leave_repo.delete_many([row.id for row in existing]) if day_type == DAY_TYPE_NONE: return now = datetime.now() await self._planned_leave_repo.create( PlannedLeaveDay( id=uuid4(), workspace_id=workspace_id, child_id=child_id, parent_personal_data_uuid=parent_personal_data_uuid, date=day, day_type=day_type, is_dubbeldag=is_dubbeldag, wish_id=wish_id, created_at=now, updated_at=now, ) ) async def clear( self, workspace_id: UUID, child_id: UUID, parent_personal_data_uuid: UUID, day: date, ) -> None: """Retract the assertion at this slot, if any (a no-op when none exists), and delete the mirrored `planned_leave_day` row at the same key, if any. Used for: un-pinning a hand-added day entirely (the Advanced modal's plain delete), deleting an ad-hoc band's days, and restoring a `day_type='none'` exclusion. """ async with self._uow(): await self._repo.delete_by_natural_key( workspace_id, child_id, parent_personal_data_uuid, day ) if self._planned_leave_repo is None: return existing = [ row for row in await self._planned_leave_repo.list_by_workspace_child( workspace_id, child_id, from_date=day, to_date=day ) if row.parent_personal_data_uuid == parent_personal_data_uuid ] if existing: await self._planned_leave_repo.delete_many([row.id for row in existing]) async def get(self, assertion_id: UUID, workspace_id: UUID | None = None) -> DayAssertion: entity = await self._repo.get_by_id(assertion_id) if entity is None or (workspace_id is not None and entity.workspace_id != workspace_id): raise NotFoundError(f"DayAssertion {assertion_id} not found") return entity async def list_for_workspace(self, workspace_id: UUID) -> list[DayAssertion]: return await self._repo.list_by_workspace(workspace_id)