"""Calendar-related Pydantic schemas. BE-CAL-004: MinimumWorkload read/write schemas. BE-CAL-API-001: TimeSlot create / response schemas. BE-CAL-API-002: Calendar day-view query schemas. BE-CAL-API-003: TimeSlot edit schemas. BE-CAL-API-004: TimeSlot cancel schemas. """ from __future__ import annotations from datetime import date, time, datetime from enum import Enum from pydantic import BaseModel, Field, model_validator, field_validator from typing import Any, Optional # --------------------------------------------------------------------------- # MinimumWorkload # --------------------------------------------------------------------------- class WorkloadCategoryThresholds(BaseModel): """Minutes thresholds per slot category within a single period.""" work: int = Field(0, ge=0, le=65535, description="Minutes of work-type slots") on_call: int = Field(0, ge=0, le=65535, description="Minutes of on-call-type slots") entertainment: int = Field(0, ge=0, le=65535, description="Minutes of entertainment-type slots") class MinimumWorkloadConfig(BaseModel): """Full workload configuration across all four periods.""" daily: WorkloadCategoryThresholds = Field(default_factory=WorkloadCategoryThresholds) weekly: WorkloadCategoryThresholds = Field(default_factory=WorkloadCategoryThresholds) monthly: WorkloadCategoryThresholds = Field(default_factory=WorkloadCategoryThresholds) yearly: WorkloadCategoryThresholds = Field(default_factory=WorkloadCategoryThresholds) class MinimumWorkloadUpdate(BaseModel): """Partial update — only provided periods/categories are overwritten. Accepts the same shape as ``MinimumWorkloadConfig`` but every field is optional so callers can PATCH individual periods. """ daily: Optional[WorkloadCategoryThresholds] = None weekly: Optional[WorkloadCategoryThresholds] = None monthly: Optional[WorkloadCategoryThresholds] = None yearly: Optional[WorkloadCategoryThresholds] = None class MinimumWorkloadResponse(BaseModel): """API response for workload configuration.""" user_id: int config: MinimumWorkloadConfig class Config: from_attributes = True # --------------------------------------------------------------------------- # Workload warning (used by future calendar validation endpoints) # --------------------------------------------------------------------------- class WorkloadWarningItem(BaseModel): """A single workload warning returned alongside a calendar mutation.""" period: str = Field(..., description="daily | weekly | monthly | yearly") category: str = Field(..., description="work | on_call | entertainment") current_minutes: int = Field(..., ge=0, description="Current scheduled minutes in the period") minimum_minutes: int = Field(..., ge=0, description="Configured minimum threshold") shortfall_minutes: int = Field(..., ge=0, description="How many minutes below threshold") message: str = Field(..., description="Human-readable warning") # --------------------------------------------------------------------------- # TimeSlot enums (mirror DB enums for schema layer) # --------------------------------------------------------------------------- class SlotTypeEnum(str, Enum): WORK = "work" ON_CALL = "on_call" ENTERTAINMENT = "entertainment" SYSTEM = "system" class EventTypeEnum(str, Enum): JOB = "job" ENTERTAINMENT = "entertainment" SYSTEM_EVENT = "system_event" class SlotStatusEnum(str, Enum): NOT_STARTED = "not_started" ONGOING = "ongoing" DEFERRED = "deferred" SKIPPED = "skipped" PAUSED = "paused" FINISHED = "finished" ABORTED = "aborted" # --------------------------------------------------------------------------- # TimeSlot create / response (BE-CAL-API-001) # --------------------------------------------------------------------------- class TimeSlotCreate(BaseModel): """Request body for creating a single calendar slot.""" date: Optional[date] = Field(None, description="Target date (defaults to today)") slot_type: SlotTypeEnum = Field(..., description="work | on_call | entertainment | system") scheduled_at: time = Field(..., description="Planned start time HH:MM (00:00-23:00)") estimated_duration: int = Field(..., ge=1, le=50, description="Duration in minutes (1-50)") event_type: Optional[EventTypeEnum] = Field(None, description="job | entertainment | system_event") event_data: Optional[dict[str, Any]] = Field(None, description="Event details JSON") priority: int = Field(0, ge=0, le=99, description="Priority 0-99") @field_validator("scheduled_at") @classmethod def _validate_scheduled_at(cls, v: time) -> time: if v.hour > 23: raise ValueError("scheduled_at hour must be between 00 and 23") return v class SlotConflictItem(BaseModel): """Describes a single overlap conflict.""" conflicting_slot_id: Optional[int] = None conflicting_virtual_id: Optional[str] = None scheduled_at: str estimated_duration: int slot_type: str message: str class TimeSlotResponse(BaseModel): """Response for a single TimeSlot.""" id: int user_id: int date: date slot_type: str estimated_duration: int scheduled_at: str # HH:MM:SS ISO format started_at: Optional[str] = None attended: bool actual_duration: Optional[int] = None event_type: Optional[str] = None event_data: Optional[dict[str, Any]] = None priority: int status: str plan_id: Optional[int] = None created_at: Optional[datetime] = None updated_at: Optional[datetime] = None class Config: from_attributes = True class TimeSlotCreateResponse(BaseModel): """Response after creating a slot — includes the slot and any warnings.""" slot: TimeSlotResponse warnings: list[WorkloadWarningItem] = Field(default_factory=list) # --------------------------------------------------------------------------- # TimeSlot edit (BE-CAL-API-003) # --------------------------------------------------------------------------- class TimeSlotEdit(BaseModel): """Request body for editing a calendar slot. All fields are optional — only provided fields are updated. The caller must supply either ``slot_id`` (for real slots) or ``virtual_id`` (for plan-generated virtual slots) in the URL path. """ slot_type: Optional[SlotTypeEnum] = Field(None, description="New slot type") scheduled_at: Optional[time] = Field(None, description="New start time HH:MM") estimated_duration: Optional[int] = Field(None, ge=1, le=50, description="New duration in minutes (1-50)") event_type: Optional[EventTypeEnum] = Field(None, description="New event type") event_data: Optional[dict[str, Any]] = Field(None, description="New event details JSON") priority: Optional[int] = Field(None, ge=0, le=99, description="New priority 0-99") @field_validator("scheduled_at") @classmethod def _validate_scheduled_at(cls, v: Optional[time]) -> Optional[time]: if v is not None and v.hour > 23: raise ValueError("scheduled_at hour must be between 00 and 23") return v @model_validator(mode="after") def _at_least_one_field(self) -> "TimeSlotEdit": """Ensure at least one editable field is provided.""" if all( getattr(self, f) is None for f in ("slot_type", "scheduled_at", "estimated_duration", "event_type", "event_data", "priority") ): raise ValueError("At least one field must be provided for edit") return self class TimeSlotEditResponse(BaseModel): """Response after editing a slot — includes the updated slot and any warnings.""" slot: TimeSlotResponse warnings: list[WorkloadWarningItem] = Field(default_factory=list) # --------------------------------------------------------------------------- # Calendar day-view query (BE-CAL-API-002) # --------------------------------------------------------------------------- class CalendarSlotItem(BaseModel): """Unified slot item for day-view — covers both real and virtual slots. * For **real** (materialized) slots: ``id`` is set, ``virtual_id`` is None. * For **virtual** (plan-generated) slots: ``id`` is None, ``virtual_id`` is the ``plan-{plan_id}-{date}`` identifier. """ id: Optional[int] = Field(None, description="Real slot DB id (None for virtual)") virtual_id: Optional[str] = Field(None, description="Virtual slot id (None for real)") user_id: int date: date slot_type: str estimated_duration: int scheduled_at: str # HH:MM:SS ISO format started_at: Optional[str] = None attended: bool actual_duration: Optional[int] = None event_type: Optional[str] = None event_data: Optional[dict[str, Any]] = None priority: int status: str plan_id: Optional[int] = None created_at: Optional[datetime] = None updated_at: Optional[datetime] = None class Config: from_attributes = True class CalendarDayResponse(BaseModel): """Response for a single-day calendar query.""" date: date user_id: int slots: list[CalendarSlotItem] = Field( default_factory=list, description="All slots for the day, sorted by scheduled_at ascending", ) # --------------------------------------------------------------------------- # TimeSlot cancel (BE-CAL-API-004) # --------------------------------------------------------------------------- class TimeSlotCancelResponse(BaseModel): """Response after cancelling a slot — includes the cancelled slot.""" slot: TimeSlotResponse message: str = Field("Slot cancelled successfully", description="Human-readable result")