zhi
78d836c71e
- PATCH /calendar/plans/{plan_id}: edit a recurring schedule plan
- Validates period-parameter hierarchy after merge
- Rejects edits to inactive (cancelled) plans
- Detaches future materialized slots so they keep old data
- Past materialized slots remain untouched
- POST /calendar/plans/{plan_id}/cancel: cancel (soft-delete) a plan
- Sets is_active=False
- Detaches future materialized slots (plan_id -> NULL)
- Preserves past materialized slots, returns their IDs
- Added SchedulePlanEdit and SchedulePlanCancelResponse schemas
394 lines
15 KiB
Python
394 lines
15 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.
|
|
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")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SchedulePlan enums (mirror DB enums)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class DayOfWeekEnum(str, Enum):
|
|
SUN = "sun"
|
|
MON = "mon"
|
|
TUE = "tue"
|
|
WED = "wed"
|
|
THU = "thu"
|
|
FRI = "fri"
|
|
SAT = "sat"
|
|
|
|
|
|
class MonthOfYearEnum(str, Enum):
|
|
JAN = "jan"
|
|
FEB = "feb"
|
|
MAR = "mar"
|
|
APR = "apr"
|
|
MAY = "may"
|
|
JUN = "jun"
|
|
JUL = "jul"
|
|
AUG = "aug"
|
|
SEP = "sep"
|
|
OCT = "oct"
|
|
NOV = "nov"
|
|
DEC = "dec"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SchedulePlan create / response (BE-CAL-API-005)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class SchedulePlanCreate(BaseModel):
|
|
"""Request body for creating a recurring schedule plan."""
|
|
slot_type: SlotTypeEnum = Field(..., description="work | on_call | entertainment | system")
|
|
estimated_duration: int = Field(..., ge=1, le=50, description="Duration in minutes (1-50)")
|
|
at_time: time = Field(..., description="Daily scheduled time (HH:MM)")
|
|
on_day: Optional[DayOfWeekEnum] = Field(None, description="Day of week (sun-sat)")
|
|
on_week: Optional[int] = Field(None, ge=1, le=4, description="Week of month (1-4)")
|
|
on_month: Optional[MonthOfYearEnum] = Field(None, description="Month (jan-dec)")
|
|
event_type: Optional[EventTypeEnum] = Field(None, description="job | entertainment | system_event")
|
|
event_data: Optional[dict[str, Any]] = Field(None, description="Event details JSON")
|
|
|
|
@field_validator("at_time")
|
|
@classmethod
|
|
def _validate_at_time(cls, v: time) -> time:
|
|
if v.hour > 23:
|
|
raise ValueError("at_time hour must be between 00 and 23")
|
|
return v
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_hierarchy(self) -> "SchedulePlanCreate":
|
|
"""Enforce period-parameter hierarchy: on_month → on_week → on_day."""
|
|
if self.on_month is not None and self.on_week is None:
|
|
raise ValueError("on_month requires on_week to be set")
|
|
if self.on_week is not None and self.on_day is None:
|
|
raise ValueError("on_week requires on_day to be set")
|
|
return self
|
|
|
|
|
|
class SchedulePlanResponse(BaseModel):
|
|
"""Response for a single SchedulePlan."""
|
|
id: int
|
|
user_id: int
|
|
slot_type: str
|
|
estimated_duration: int
|
|
at_time: str # HH:MM:SS ISO format
|
|
on_day: Optional[str] = None
|
|
on_week: Optional[int] = None
|
|
on_month: Optional[str] = None
|
|
event_type: Optional[str] = None
|
|
event_data: Optional[dict[str, Any]] = None
|
|
is_active: bool
|
|
created_at: Optional[datetime] = None
|
|
updated_at: Optional[datetime] = None
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class SchedulePlanListResponse(BaseModel):
|
|
"""Response for listing schedule plans."""
|
|
plans: list[SchedulePlanResponse] = Field(default_factory=list)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SchedulePlan edit / cancel (BE-CAL-API-006)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class SchedulePlanEdit(BaseModel):
|
|
"""Request body for editing a recurring schedule plan.
|
|
|
|
All fields are optional — only provided fields are updated.
|
|
Period-parameter hierarchy (on_month → on_week → on_day) is
|
|
validated after merging with existing plan values.
|
|
"""
|
|
slot_type: Optional[SlotTypeEnum] = Field(None, description="New slot type")
|
|
estimated_duration: Optional[int] = Field(None, ge=1, le=50, description="New duration in minutes (1-50)")
|
|
at_time: Optional[time] = Field(None, description="New daily time (HH:MM)")
|
|
on_day: Optional[DayOfWeekEnum] = Field(None, description="New day of week (sun-sat), use 'clear' param to remove")
|
|
on_week: Optional[int] = Field(None, ge=1, le=4, description="New week of month (1-4), use 'clear' param to remove")
|
|
on_month: Optional[MonthOfYearEnum] = Field(None, description="New month (jan-dec), use 'clear' param to remove")
|
|
event_type: Optional[EventTypeEnum] = Field(None, description="New event type")
|
|
event_data: Optional[dict[str, Any]] = Field(None, description="New event details JSON")
|
|
clear_on_day: bool = Field(False, description="Clear on_day (set to NULL)")
|
|
clear_on_week: bool = Field(False, description="Clear on_week (set to NULL)")
|
|
clear_on_month: bool = Field(False, description="Clear on_month (set to NULL)")
|
|
|
|
@field_validator("at_time")
|
|
@classmethod
|
|
def _validate_at_time(cls, v: Optional[time]) -> Optional[time]:
|
|
if v is not None and v.hour > 23:
|
|
raise ValueError("at_time hour must be between 00 and 23")
|
|
return v
|
|
|
|
@model_validator(mode="after")
|
|
def _at_least_one_field(self) -> "SchedulePlanEdit":
|
|
"""Ensure at least one editable field or clear flag is provided."""
|
|
has_value = any(
|
|
getattr(self, f) is not None
|
|
for f in ("slot_type", "estimated_duration", "at_time", "on_day",
|
|
"on_week", "on_month", "event_type", "event_data")
|
|
)
|
|
has_clear = self.clear_on_day or self.clear_on_week or self.clear_on_month
|
|
if not has_value and not has_clear:
|
|
raise ValueError("At least one field must be provided for edit")
|
|
return self
|
|
|
|
|
|
class SchedulePlanCancelResponse(BaseModel):
|
|
"""Response after cancelling a plan."""
|
|
plan: SchedulePlanResponse
|
|
message: str = Field("Plan cancelled successfully", description="Human-readable result")
|
|
preserved_past_slot_ids: list[int] = Field(
|
|
default_factory=list,
|
|
description="IDs of past materialized slots that were NOT affected",
|
|
)
|