- Add GET /calendar/day endpoint with optional ?date= query param - Returns unified CalendarDayResponse merging real slots + virtual plan slots - New CalendarSlotItem schema supports both real (id) and virtual (virtual_id) slots - Excludes inactive slots (skipped/aborted) from results - All slots sorted by scheduled_at ascending - Helper functions for real/virtual slot conversion
199 lines
7.2 KiB
Python
199 lines
7.2 KiB
Python
"""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.
|
|
"""
|
|
|
|
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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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",
|
|
)
|