HarborForge.Backend: dev-2026-03-29 -> main #13
@@ -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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user