From 751b3bc5744e7af73afdd03dddb883e2666bc316 Mon Sep 17 00:00:00 2001 From: zhi Date: Tue, 31 Mar 2026 05:45:58 +0000 Subject: [PATCH] BE-CAL-API-001: Implement single slot creation API - Add TimeSlotCreate, TimeSlotResponse, TimeSlotCreateResponse schemas - Add SlotConflictItem, SlotTypeEnum, EventTypeEnum, SlotStatusEnum to schemas - Add POST /calendar/slots endpoint with overlap detection and workload warnings - Add _slot_to_response helper for ORM -> schema conversion --- app/api/routers/calendar.py | 99 ++++++++++++++++++++++++++++++++++++- app/schemas/calendar.py | 95 ++++++++++++++++++++++++++++++++++- 2 files changed, 191 insertions(+), 3 deletions(-) diff --git a/app/api/routers/calendar.py b/app/api/routers/calendar.py index a6d8790..eb05f91 100644 --- a/app/api/routers/calendar.py +++ b/app/api/routers/calendar.py @@ -1,29 +1,126 @@ """Calendar API router. BE-CAL-004: MinimumWorkload CRUD endpoints. -Future tasks (BE-CAL-API-*) will add slot/plan endpoints here. +BE-CAL-API-001: Single-slot creation endpoint. """ +from datetime import date as date_type + from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.api.deps import get_current_user from app.core.config import get_db +from app.models.calendar import SlotStatus, TimeSlot from app.models.models import User from app.schemas.calendar import ( MinimumWorkloadConfig, MinimumWorkloadResponse, MinimumWorkloadUpdate, + SlotConflictItem, + TimeSlotCreate, + TimeSlotCreateResponse, + TimeSlotResponse, ) from app.services.minimum_workload import ( get_workload_config, + get_workload_warnings_for_date, replace_workload_config, upsert_workload_config, ) +from app.services.overlap import check_overlap_for_create router = APIRouter(prefix="/calendar", tags=["Calendar"]) +# --------------------------------------------------------------------------- +# TimeSlot creation (BE-CAL-API-001) +# --------------------------------------------------------------------------- + +def _slot_to_response(slot: TimeSlot) -> TimeSlotResponse: + """Convert a TimeSlot ORM object to a response schema.""" + return TimeSlotResponse( + id=slot.id, + user_id=slot.user_id, + date=slot.date, + slot_type=slot.slot_type.value if hasattr(slot.slot_type, "value") else str(slot.slot_type), + estimated_duration=slot.estimated_duration, + scheduled_at=slot.scheduled_at.isoformat() if slot.scheduled_at else "", + started_at=slot.started_at.isoformat() if slot.started_at else None, + attended=slot.attended, + actual_duration=slot.actual_duration, + event_type=slot.event_type.value if slot.event_type and hasattr(slot.event_type, "value") else (str(slot.event_type) if slot.event_type else None), + event_data=slot.event_data, + priority=slot.priority, + status=slot.status.value if hasattr(slot.status, "value") else str(slot.status), + plan_id=slot.plan_id, + created_at=slot.created_at, + updated_at=slot.updated_at, + ) + + +@router.post( + "/slots", + response_model=TimeSlotCreateResponse, + status_code=201, + summary="Create a single calendar slot", +) +def create_slot( + payload: TimeSlotCreate, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """Create a one-off calendar slot. + + - **Overlap detection**: rejects the request if the proposed slot + overlaps with existing real or virtual slots on the same day. + - **Workload warnings**: after successful creation, returns any + minimum-workload warnings (advisory only, does not block creation). + """ + target_date = payload.date or date_type.today() + + # --- Overlap check (hard reject) --- + conflicts = check_overlap_for_create( + db, + user_id=current_user.id, + target_date=target_date, + scheduled_at=payload.scheduled_at, + estimated_duration=payload.estimated_duration, + ) + if conflicts: + raise HTTPException( + status_code=409, + detail={ + "message": "Slot overlaps with existing schedule", + "conflicts": [c.to_dict() for c in conflicts], + }, + ) + + # --- Create the slot --- + slot = TimeSlot( + user_id=current_user.id, + date=target_date, + slot_type=payload.slot_type.value, + estimated_duration=payload.estimated_duration, + scheduled_at=payload.scheduled_at, + event_type=payload.event_type.value if payload.event_type else None, + event_data=payload.event_data, + priority=payload.priority, + status=SlotStatus.NOT_STARTED, + ) + db.add(slot) + db.commit() + db.refresh(slot) + + # --- Workload warnings (advisory) --- + warnings = get_workload_warnings_for_date(db, current_user.id, target_date) + + return TimeSlotCreateResponse( + slot=_slot_to_response(slot), + warnings=warnings, + ) + + # --------------------------------------------------------------------------- # MinimumWorkload # --------------------------------------------------------------------------- diff --git a/app/schemas/calendar.py b/app/schemas/calendar.py index 1ad97c8..8e59ec3 100644 --- a/app/schemas/calendar.py +++ b/app/schemas/calendar.py @@ -1,12 +1,15 @@ """Calendar-related Pydantic schemas. BE-CAL-004: MinimumWorkload read/write schemas. +BE-CAL-API-001: TimeSlot create / response schemas. """ from __future__ import annotations -from pydantic import BaseModel, Field, model_validator -from typing import Optional +from datetime import date, time, datetime +from enum import Enum +from pydantic import BaseModel, Field, model_validator, field_validator +from typing import Any, Optional # --------------------------------------------------------------------------- @@ -61,3 +64,91 @@ class WorkloadWarningItem(BaseModel): 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)