HarborForge.Backend: dev-2026-03-29 -> main #13

Merged
hzhang merged 43 commits from dev-2026-03-29 into main 2026-04-05 22:08:15 +00:00
2 changed files with 244 additions and 2 deletions
Showing only changes of commit f7f9ba3aa7 - Show all commits

View File

@@ -3,6 +3,7 @@
BE-CAL-004: MinimumWorkload CRUD endpoints.
BE-CAL-API-001: Single-slot creation endpoint.
BE-CAL-API-002: Day-view calendar query endpoint.
BE-CAL-API-003: Calendar slot edit endpoints (real + virtual).
"""
from datetime import date as date_type
@@ -24,6 +25,8 @@ from app.schemas.calendar import (
SlotConflictItem,
TimeSlotCreate,
TimeSlotCreateResponse,
TimeSlotEdit,
TimeSlotEditResponse,
TimeSlotResponse,
)
from app.services.minimum_workload import (
@@ -32,8 +35,18 @@ from app.services.minimum_workload import (
replace_workload_config,
upsert_workload_config,
)
from app.services.overlap import check_overlap_for_create
from app.services.plan_slot import get_virtual_slots_for_date
from app.services.overlap import check_overlap_for_create, check_overlap_for_edit
from app.services.plan_slot import (
detach_slot_from_plan,
get_virtual_slots_for_date,
materialize_from_virtual_id,
parse_virtual_slot_id,
)
from app.services.slot_immutability import (
ImmutableSlotError,
guard_edit_real_slot,
guard_edit_virtual_slot,
)
router = APIRouter(prefix="/calendar", tags=["Calendar"])
@@ -244,6 +257,191 @@ def get_calendar_day(
)
# ---------------------------------------------------------------------------
# Slot edit (BE-CAL-API-003)
# ---------------------------------------------------------------------------
def _apply_edit_fields(slot: TimeSlot, payload: TimeSlotEdit) -> None:
"""Apply non-None fields from *payload* to a TimeSlot ORM object."""
if payload.slot_type is not None:
slot.slot_type = payload.slot_type.value
if payload.scheduled_at is not None:
slot.scheduled_at = payload.scheduled_at
if payload.estimated_duration is not None:
slot.estimated_duration = payload.estimated_duration
if payload.event_type is not None:
slot.event_type = payload.event_type.value
if payload.event_data is not None:
slot.event_data = payload.event_data
if payload.priority is not None:
slot.priority = payload.priority
@router.patch(
"/slots/{slot_id}",
response_model=TimeSlotEditResponse,
summary="Edit a real (materialized) calendar slot",
)
def edit_real_slot(
slot_id: int,
payload: TimeSlotEdit,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Edit an existing real (materialized) slot.
- **Immutability**: rejects edits to past slots.
- **Overlap detection**: if time/duration changed, rejects on overlap
(excluding the slot being edited).
- **Plan detach**: if the slot was materialized from a plan, editing
detaches it (sets ``plan_id`` to NULL).
- **Workload warnings**: returned after successful edit (advisory only).
"""
slot = (
db.query(TimeSlot)
.filter(TimeSlot.id == slot_id, TimeSlot.user_id == current_user.id)
.first()
)
if slot is None:
raise HTTPException(status_code=404, detail="Slot not found")
# --- Past-slot guard ---
try:
guard_edit_real_slot(db, slot)
except ImmutableSlotError as e:
raise HTTPException(status_code=422, detail=str(e))
# --- Determine effective time/duration for overlap check ---
effective_scheduled_at = payload.scheduled_at if payload.scheduled_at is not None else slot.scheduled_at
effective_duration = payload.estimated_duration if payload.estimated_duration is not None else slot.estimated_duration
# --- Overlap check (if time or duration changed) ---
time_changed = (
(payload.scheduled_at is not None and payload.scheduled_at != slot.scheduled_at)
or (payload.estimated_duration is not None and payload.estimated_duration != slot.estimated_duration)
)
if time_changed:
conflicts = check_overlap_for_edit(
db,
user_id=current_user.id,
slot_id=slot.id,
target_date=slot.date,
scheduled_at=effective_scheduled_at,
estimated_duration=effective_duration,
)
if conflicts:
raise HTTPException(
status_code=409,
detail={
"message": "Edited slot overlaps with existing schedule",
"conflicts": [c.to_dict() for c in conflicts],
},
)
# --- Detach from plan if applicable ---
if slot.plan_id is not None:
detach_slot_from_plan(slot)
# --- Apply edits ---
_apply_edit_fields(slot, payload)
db.commit()
db.refresh(slot)
# --- Workload warnings ---
warnings = get_workload_warnings_for_date(db, current_user.id, slot.date)
return TimeSlotEditResponse(
slot=_slot_to_response(slot),
warnings=warnings,
)
@router.patch(
"/slots/virtual/{virtual_id}",
response_model=TimeSlotEditResponse,
summary="Edit a virtual (plan-generated) calendar slot",
)
def edit_virtual_slot(
virtual_id: str,
payload: TimeSlotEdit,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Edit a virtual (plan-generated) slot.
This triggers **materialization**: the virtual slot is first converted
into a real TimeSlot row, then the edits are applied, and the slot is
detached from its plan (``plan_id`` set to NULL).
- **Immutability**: rejects edits to past virtual slots.
- **Overlap detection**: checks overlap with the edited time/duration
against existing slots on the same day.
- **Workload warnings**: returned after successful edit (advisory only).
"""
# --- Validate virtual_id format ---
parsed = parse_virtual_slot_id(virtual_id)
if parsed is None:
raise HTTPException(status_code=400, detail=f"Invalid virtual slot id: {virtual_id}")
plan_id, slot_date = parsed
# --- Past-slot guard ---
try:
guard_edit_virtual_slot(virtual_id)
except ImmutableSlotError as e:
raise HTTPException(status_code=422, detail=str(e))
# --- Materialize ---
try:
slot = materialize_from_virtual_id(db, virtual_id)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
# --- Verify ownership ---
if slot.user_id != current_user.id:
db.rollback()
raise HTTPException(status_code=404, detail="Slot not found")
# --- Determine effective time/duration for overlap check ---
effective_scheduled_at = payload.scheduled_at if payload.scheduled_at is not None else slot.scheduled_at
effective_duration = payload.estimated_duration if payload.estimated_duration is not None else slot.estimated_duration
# --- Overlap check (exclude newly materialized slot) ---
conflicts = check_overlap_for_edit(
db,
user_id=current_user.id,
slot_id=slot.id,
target_date=slot.date,
scheduled_at=effective_scheduled_at,
estimated_duration=effective_duration,
)
if conflicts:
db.rollback()
raise HTTPException(
status_code=409,
detail={
"message": "Edited slot overlaps with existing schedule",
"conflicts": [c.to_dict() for c in conflicts],
},
)
# --- Detach from plan ---
detach_slot_from_plan(slot)
# --- Apply edits ---
_apply_edit_fields(slot, payload)
db.commit()
db.refresh(slot)
# --- Workload warnings ---
warnings = get_workload_warnings_for_date(db, current_user.id, slot.date)
return TimeSlotEditResponse(
slot=_slot_to_response(slot),
warnings=warnings,
)
# ---------------------------------------------------------------------------
# MinimumWorkload
# ---------------------------------------------------------------------------

View File

@@ -3,6 +3,7 @@
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.
"""
from __future__ import annotations
@@ -155,6 +156,49 @@ class TimeSlotCreateResponse(BaseModel):
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)
# ---------------------------------------------------------------------------