Piotr Oleszczyk
9bf94a979c
- Rename skincare route prefix /skin-snapshots → /skincare to match API client - Add redirect_slashes=False to FastAPI app; change collection routes from "/" to "" to eliminate 307 redirects on POST/GET without trailing slash - Fix redirect() inside try/catch in products/new and routines/new server actions (SvelteKit redirect() throws and was being caught as a 500 error) - Eagerly load inventory and steps relationships via explicit SELECT + model_dump(mode="json"), working around SQLModel 0.0.37 not serializing Relationship fields in response_model - Add field_validator for product_effect_profile to coerce DB-returned dict → ProductEffectProfile, eliminating Pydantic serializer warning - Update all tests to use routes without trailing slash Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
133 lines
4.1 KiB
Python
133 lines
4.1 KiB
Python
from datetime import date
|
|
from typing import Optional
|
|
from uuid import UUID, uuid4
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlmodel import Session, SQLModel, select
|
|
|
|
from db import get_session
|
|
from innercontext.models import SkinConditionSnapshot
|
|
from innercontext.models.enums import (
|
|
BarrierState,
|
|
OverallSkinState,
|
|
SkinConcern,
|
|
SkinTrend,
|
|
SkinType,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class SnapshotCreate(SQLModel):
|
|
snapshot_date: date
|
|
overall_state: Optional[OverallSkinState] = None
|
|
trend: Optional[SkinTrend] = None
|
|
skin_type: Optional[SkinType] = None
|
|
|
|
hydration_level: Optional[int] = None
|
|
sebum_tzone: Optional[int] = None
|
|
sebum_cheeks: Optional[int] = None
|
|
sensitivity_level: Optional[int] = None
|
|
|
|
barrier_state: Optional[BarrierState] = None
|
|
|
|
active_concerns: list[SkinConcern] = []
|
|
risks: list[str] = []
|
|
priorities: list[str] = []
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class SnapshotUpdate(SQLModel):
|
|
snapshot_date: Optional[date] = None
|
|
overall_state: Optional[OverallSkinState] = None
|
|
trend: Optional[SkinTrend] = None
|
|
skin_type: Optional[SkinType] = None
|
|
|
|
hydration_level: Optional[int] = None
|
|
sebum_tzone: Optional[int] = None
|
|
sebum_cheeks: Optional[int] = None
|
|
sensitivity_level: Optional[int] = None
|
|
|
|
barrier_state: Optional[BarrierState] = None
|
|
|
|
active_concerns: Optional[list[SkinConcern]] = None
|
|
risks: Optional[list[str]] = None
|
|
priorities: Optional[list[str]] = None
|
|
notes: Optional[str] = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def get_or_404(session: Session, model, record_id) -> object:
|
|
obj = session.get(model, record_id)
|
|
if obj is None:
|
|
raise HTTPException(status_code=404, detail=f"{model.__name__} not found")
|
|
return obj
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("", response_model=list[SkinConditionSnapshot])
|
|
def list_snapshots(
|
|
from_date: Optional[date] = None,
|
|
to_date: Optional[date] = None,
|
|
overall_state: Optional[OverallSkinState] = None,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
stmt = select(SkinConditionSnapshot)
|
|
if from_date is not None:
|
|
stmt = stmt.where(SkinConditionSnapshot.snapshot_date >= from_date)
|
|
if to_date is not None:
|
|
stmt = stmt.where(SkinConditionSnapshot.snapshot_date <= to_date)
|
|
if overall_state is not None:
|
|
stmt = stmt.where(SkinConditionSnapshot.overall_state == overall_state)
|
|
return session.exec(stmt).all()
|
|
|
|
|
|
@router.post("", response_model=SkinConditionSnapshot, status_code=201)
|
|
def create_snapshot(
|
|
data: SnapshotCreate, session: Session = Depends(get_session)
|
|
):
|
|
snapshot = SkinConditionSnapshot(id=uuid4(), **data.model_dump())
|
|
session.add(snapshot)
|
|
session.commit()
|
|
session.refresh(snapshot)
|
|
return snapshot
|
|
|
|
|
|
@router.get("/{snapshot_id}", response_model=SkinConditionSnapshot)
|
|
def get_snapshot(snapshot_id: UUID, session: Session = Depends(get_session)):
|
|
return get_or_404(session, SkinConditionSnapshot, snapshot_id)
|
|
|
|
|
|
@router.patch("/{snapshot_id}", response_model=SkinConditionSnapshot)
|
|
def update_snapshot(
|
|
snapshot_id: UUID,
|
|
data: SnapshotUpdate,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
snapshot = get_or_404(session, SkinConditionSnapshot, snapshot_id)
|
|
for key, value in data.model_dump(exclude_unset=True).items():
|
|
setattr(snapshot, key, value)
|
|
session.add(snapshot)
|
|
session.commit()
|
|
session.refresh(snapshot)
|
|
return snapshot
|
|
|
|
|
|
@router.delete("/{snapshot_id}", status_code=204)
|
|
def delete_snapshot(snapshot_id: UUID, session: Session = Depends(get_session)):
|
|
snapshot = get_or_404(session, SkinConditionSnapshot, snapshot_id)
|
|
session.delete(snapshot)
|
|
session.commit()
|