Initial commit: backend API, data models, and test suite

FastAPI backend for personal health and skincare data with MCP export.
Includes SQLModel models for products, inventory, medications, lab results,
routines, and skin condition snapshots. Pytest suite with 111 tests running
on SQLite in-memory (no PostgreSQL required).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Piotr Oleszczyk 2026-02-26 15:10:24 +01:00
commit 8f7d893a63
32 changed files with 6282 additions and 0 deletions

View file

View file

@ -0,0 +1,317 @@
import re
from datetime import datetime
from typing import Optional
from uuid import UUID, uuid4
from fastapi import APIRouter, Depends, HTTPException
from pydantic import field_validator
from sqlmodel import Session, SQLModel, select
from db import get_session
from innercontext.models import LabResult, MedicationEntry, MedicationUsage
from innercontext.models.enums import MedicationKind, ResultFlag
router = APIRouter()
# ---------------------------------------------------------------------------
# Schemas
# ---------------------------------------------------------------------------
class MedicationCreate(SQLModel):
kind: MedicationKind
product_name: str
active_substance: Optional[str] = None
formulation: Optional[str] = None
route: Optional[str] = None
source_file: Optional[str] = None
notes: Optional[str] = None
class MedicationUpdate(SQLModel):
kind: Optional[MedicationKind] = None
product_name: Optional[str] = None
active_substance: Optional[str] = None
formulation: Optional[str] = None
route: Optional[str] = None
source_file: Optional[str] = None
notes: Optional[str] = None
class UsageCreate(SQLModel):
dose_value: Optional[float] = None
dose_unit: Optional[str] = None
frequency: Optional[str] = None
schedule_text: Optional[str] = None
as_needed: bool = False
valid_from: datetime
valid_to: Optional[datetime] = None
source_file: Optional[str] = None
notes: Optional[str] = None
class UsageUpdate(SQLModel):
dose_value: Optional[float] = None
dose_unit: Optional[str] = None
frequency: Optional[str] = None
schedule_text: Optional[str] = None
as_needed: Optional[bool] = None
valid_from: Optional[datetime] = None
valid_to: Optional[datetime] = None
source_file: Optional[str] = None
notes: Optional[str] = None
class LabResultCreate(SQLModel):
collected_at: datetime
test_code: str
@field_validator("test_code")
@classmethod
def check_test_code_format(cls, v: str) -> str:
if not re.fullmatch(r"\d+-\d", v):
raise ValueError("test_code must match LOINC format: digits-digit (e.g. 718-7)")
return v
test_name_original: Optional[str] = None
test_name_loinc: Optional[str] = None
value_num: Optional[float] = None
value_text: Optional[str] = None
value_bool: Optional[bool] = None
unit_original: Optional[str] = None
unit_ucum: Optional[str] = None
ref_low: Optional[float] = None
ref_high: Optional[float] = None
ref_text: Optional[str] = None
flag: Optional[ResultFlag] = None
lab: Optional[str] = None
source_file: Optional[str] = None
notes: Optional[str] = None
class LabResultUpdate(SQLModel):
collected_at: Optional[datetime] = None
test_code: Optional[str] = None
test_name_original: Optional[str] = None
test_name_loinc: Optional[str] = None
value_num: Optional[float] = None
value_text: Optional[str] = None
value_bool: Optional[bool] = None
unit_original: Optional[str] = None
unit_ucum: Optional[str] = None
ref_low: Optional[float] = None
ref_high: Optional[float] = None
ref_text: Optional[str] = None
flag: Optional[ResultFlag] = None
lab: Optional[str] = None
source_file: Optional[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
# ---------------------------------------------------------------------------
# Medication routes
# ---------------------------------------------------------------------------
@router.get("/medications", response_model=list[MedicationEntry])
def list_medications(
kind: Optional[MedicationKind] = None,
product_name: Optional[str] = None,
session: Session = Depends(get_session),
):
stmt = select(MedicationEntry)
if kind is not None:
stmt = stmt.where(MedicationEntry.kind == kind)
if product_name is not None:
stmt = stmt.where(MedicationEntry.product_name.ilike(f"%{product_name}%"))
return session.exec(stmt).all()
@router.post("/medications", response_model=MedicationEntry, status_code=201)
def create_medication(
data: MedicationCreate, session: Session = Depends(get_session)
):
entry = MedicationEntry(record_id=uuid4(), **data.model_dump())
session.add(entry)
session.commit()
session.refresh(entry)
return entry
@router.get("/medications/{medication_id}", response_model=MedicationEntry)
def get_medication(medication_id: UUID, session: Session = Depends(get_session)):
return get_or_404(session, MedicationEntry, medication_id)
@router.patch("/medications/{medication_id}", response_model=MedicationEntry)
def update_medication(
medication_id: UUID,
data: MedicationUpdate,
session: Session = Depends(get_session),
):
entry = get_or_404(session, MedicationEntry, medication_id)
for key, value in data.model_dump(exclude_unset=True).items():
setattr(entry, key, value)
session.add(entry)
session.commit()
session.refresh(entry)
return entry
@router.delete("/medications/{medication_id}", status_code=204)
def delete_medication(medication_id: UUID, session: Session = Depends(get_session)):
entry = get_or_404(session, MedicationEntry, medication_id)
# Delete usages first (no cascade configured at DB level)
usages = session.exec(
select(MedicationUsage).where(
MedicationUsage.medication_record_id == medication_id
)
).all()
for u in usages:
session.delete(u)
session.delete(entry)
session.commit()
# ---------------------------------------------------------------------------
# Usage sub-routes
# ---------------------------------------------------------------------------
@router.get("/medications/{medication_id}/usages", response_model=list[MedicationUsage])
def list_usages(medication_id: UUID, session: Session = Depends(get_session)):
get_or_404(session, MedicationEntry, medication_id)
stmt = select(MedicationUsage).where(
MedicationUsage.medication_record_id == medication_id
)
return session.exec(stmt).all()
@router.post(
"/medications/{medication_id}/usages",
response_model=MedicationUsage,
status_code=201,
)
def create_usage(
medication_id: UUID,
data: UsageCreate,
session: Session = Depends(get_session),
):
get_or_404(session, MedicationEntry, medication_id)
usage = MedicationUsage(
record_id=uuid4(),
medication_record_id=medication_id,
**data.model_dump(),
)
session.add(usage)
session.commit()
session.refresh(usage)
return usage
@router.patch("/usages/{usage_id}", response_model=MedicationUsage)
def update_usage(
usage_id: UUID,
data: UsageUpdate,
session: Session = Depends(get_session),
):
usage = get_or_404(session, MedicationUsage, usage_id)
for key, value in data.model_dump(exclude_unset=True).items():
setattr(usage, key, value)
session.add(usage)
session.commit()
session.refresh(usage)
return usage
@router.delete("/usages/{usage_id}", status_code=204)
def delete_usage(usage_id: UUID, session: Session = Depends(get_session)):
usage = get_or_404(session, MedicationUsage, usage_id)
session.delete(usage)
session.commit()
# ---------------------------------------------------------------------------
# Lab result routes
# ---------------------------------------------------------------------------
@router.get("/lab-results", response_model=list[LabResult])
def list_lab_results(
test_code: Optional[str] = None,
flag: Optional[ResultFlag] = None,
lab: Optional[str] = None,
from_date: Optional[datetime] = None,
to_date: Optional[datetime] = None,
session: Session = Depends(get_session),
):
stmt = select(LabResult)
if test_code is not None:
stmt = stmt.where(LabResult.test_code == test_code)
if flag is not None:
stmt = stmt.where(LabResult.flag == flag)
if lab is not None:
stmt = stmt.where(LabResult.lab == lab)
if from_date is not None:
stmt = stmt.where(LabResult.collected_at >= from_date)
if to_date is not None:
stmt = stmt.where(LabResult.collected_at <= to_date)
return session.exec(stmt).all()
@router.post("/lab-results", response_model=LabResult, status_code=201)
def create_lab_result(
data: LabResultCreate, session: Session = Depends(get_session)
):
result = LabResult(record_id=uuid4(), **data.model_dump())
session.add(result)
session.commit()
session.refresh(result)
return result
@router.get("/lab-results/{result_id}", response_model=LabResult)
def get_lab_result(result_id: UUID, session: Session = Depends(get_session)):
return get_or_404(session, LabResult, result_id)
@router.patch("/lab-results/{result_id}", response_model=LabResult)
def update_lab_result(
result_id: UUID,
data: LabResultUpdate,
session: Session = Depends(get_session),
):
result = get_or_404(session, LabResult, result_id)
for key, value in data.model_dump(exclude_unset=True).items():
setattr(result, key, value)
session.add(result)
session.commit()
session.refresh(result)
return result
@router.delete("/lab-results/{result_id}", status_code=204)
def delete_lab_result(result_id: UUID, session: Session = Depends(get_session)):
result = get_or_404(session, LabResult, result_id)
session.delete(result)
session.commit()

View file

@ -0,0 +1,44 @@
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session
from db import get_session
from innercontext.models import ProductInventory
from innercontext.api.products import InventoryUpdate
router = APIRouter()
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
@router.get("/{inventory_id}", response_model=ProductInventory)
def get_inventory(inventory_id: UUID, session: Session = Depends(get_session)):
return get_or_404(session, ProductInventory, inventory_id)
@router.patch("/{inventory_id}", response_model=ProductInventory)
def update_inventory(
inventory_id: UUID,
data: InventoryUpdate,
session: Session = Depends(get_session),
):
entry = get_or_404(session, ProductInventory, inventory_id)
for key, value in data.model_dump(exclude_unset=True).items():
setattr(entry, key, value)
session.add(entry)
session.commit()
session.refresh(entry)
return entry
@router.delete("/{inventory_id}", status_code=204)
def delete_inventory(inventory_id: UUID, session: Session = Depends(get_session)):
entry = get_or_404(session, ProductInventory, inventory_id)
session.delete(entry)
session.commit()

View file

@ -0,0 +1,295 @@
from datetime import date
from typing import Optional
from uuid import UUID, uuid4
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlmodel import Session, SQLModel, select
from db import get_session
from innercontext.models import (
Product,
ProductCategory,
ProductInventory,
SkinConcern,
)
from innercontext.models.product import (
ActiveIngredient,
ProductContext,
ProductEffectProfile,
ProductInteraction,
)
from innercontext.models.enums import (
AbsorptionSpeed,
DayTime,
EvidenceLevel,
PriceTier,
RoutineRole,
TextureType,
UsageFrequency,
SkinType,
)
router = APIRouter()
# ---------------------------------------------------------------------------
# Request / response schemas
# ---------------------------------------------------------------------------
class ProductCreate(SQLModel):
name: str
brand: str
line_name: Optional[str] = None
sku: Optional[str] = None
url: Optional[str] = None
barcode: Optional[str] = None
category: ProductCategory
routine_role: RoutineRole
recommended_time: DayTime
texture: Optional[TextureType] = None
absorption_speed: Optional[AbsorptionSpeed] = None
leave_on: bool
price_tier: Optional[PriceTier] = None
size_ml: Optional[float] = None
pao_months: Optional[int] = None
inci: list[str] = []
actives: Optional[list[ActiveIngredient]] = None
recommended_for: list[SkinType] = []
recommended_frequency: Optional[UsageFrequency] = None
targets: list[SkinConcern] = []
contraindications: list[str] = []
usage_notes: Optional[str] = None
evidence_level: Optional[EvidenceLevel] = None
claims: list[str] = []
fragrance_free: Optional[bool] = None
essential_oils_free: Optional[bool] = None
alcohol_denat_free: Optional[bool] = None
pregnancy_safe: Optional[bool] = None
product_effect_profile: ProductEffectProfile = ProductEffectProfile()
ph_min: Optional[float] = None
ph_max: Optional[float] = None
incompatible_with: Optional[list[ProductInteraction]] = None
synergizes_with: Optional[list[str]] = None
context_rules: Optional[ProductContext] = None
min_interval_hours: Optional[int] = None
max_frequency_per_week: Optional[int] = None
is_medication: bool = False
is_tool: bool = False
needle_length_mm: Optional[float] = None
personal_rating: Optional[int] = None
personal_tolerance_notes: Optional[str] = None
personal_repurchase_intent: Optional[bool] = None
class ProductUpdate(SQLModel):
name: Optional[str] = None
brand: Optional[str] = None
line_name: Optional[str] = None
sku: Optional[str] = None
url: Optional[str] = None
barcode: Optional[str] = None
category: Optional[ProductCategory] = None
routine_role: Optional[RoutineRole] = None
recommended_time: Optional[DayTime] = None
texture: Optional[TextureType] = None
absorption_speed: Optional[AbsorptionSpeed] = None
leave_on: Optional[bool] = None
price_tier: Optional[PriceTier] = None
size_ml: Optional[float] = None
pao_months: Optional[int] = None
inci: Optional[list[str]] = None
actives: Optional[list[ActiveIngredient]] = None
recommended_for: Optional[list[SkinType]] = None
recommended_frequency: Optional[UsageFrequency] = None
targets: Optional[list[SkinConcern]] = None
contraindications: Optional[list[str]] = None
usage_notes: Optional[str] = None
evidence_level: Optional[EvidenceLevel] = None
claims: Optional[list[str]] = None
fragrance_free: Optional[bool] = None
essential_oils_free: Optional[bool] = None
alcohol_denat_free: Optional[bool] = None
pregnancy_safe: Optional[bool] = None
product_effect_profile: Optional[ProductEffectProfile] = None
ph_min: Optional[float] = None
ph_max: Optional[float] = None
incompatible_with: Optional[list[ProductInteraction]] = None
synergizes_with: Optional[list[str]] = None
context_rules: Optional[ProductContext] = None
min_interval_hours: Optional[int] = None
max_frequency_per_week: Optional[int] = None
is_medication: Optional[bool] = None
is_tool: Optional[bool] = None
needle_length_mm: Optional[float] = None
personal_rating: Optional[int] = None
personal_tolerance_notes: Optional[str] = None
personal_repurchase_intent: Optional[bool] = None
class InventoryCreate(SQLModel):
is_opened: bool = False
opened_at: Optional[date] = None
finished_at: Optional[date] = None
expiry_date: Optional[date] = None
current_weight_g: Optional[float] = None
notes: Optional[str] = None
class InventoryUpdate(SQLModel):
is_opened: Optional[bool] = None
opened_at: Optional[date] = None
finished_at: Optional[date] = None
expiry_date: Optional[date] = None
current_weight_g: Optional[float] = 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
# ---------------------------------------------------------------------------
# Product routes
# ---------------------------------------------------------------------------
@router.get("/", response_model=list[Product])
def list_products(
category: Optional[ProductCategory] = None,
brand: Optional[str] = None,
targets: Optional[list[SkinConcern]] = Query(default=None),
is_medication: Optional[bool] = None,
is_tool: Optional[bool] = None,
session: Session = Depends(get_session),
):
stmt = select(Product)
if category is not None:
stmt = stmt.where(Product.category == category)
if brand is not None:
stmt = stmt.where(Product.brand == brand)
if is_medication is not None:
stmt = stmt.where(Product.is_medication == is_medication)
if is_tool is not None:
stmt = stmt.where(Product.is_tool == is_tool)
products = session.exec(stmt).all()
# Filter by targets (JSON column — done in Python)
if targets:
target_values = {t.value for t in targets}
products = [
p
for p in products
if any(
(t.value if hasattr(t, "value") else t) in target_values
for t in (p.targets or [])
)
]
return products
@router.post("/", response_model=Product, status_code=201)
def create_product(data: ProductCreate, session: Session = Depends(get_session)):
product = Product(
id=uuid4(),
**data.model_dump(),
)
session.add(product)
session.commit()
session.refresh(product)
return product
@router.get("/{product_id}", response_model=Product)
def get_product(product_id: UUID, session: Session = Depends(get_session)):
return get_or_404(session, Product, product_id)
@router.patch("/{product_id}", response_model=Product)
def update_product(
product_id: UUID, data: ProductUpdate, session: Session = Depends(get_session)
):
product = get_or_404(session, Product, product_id)
for key, value in data.model_dump(exclude_unset=True).items():
setattr(product, key, value)
session.add(product)
session.commit()
session.refresh(product)
return product
@router.delete("/{product_id}", status_code=204)
def delete_product(product_id: UUID, session: Session = Depends(get_session)):
product = get_or_404(session, Product, product_id)
session.delete(product)
session.commit()
# ---------------------------------------------------------------------------
# Product inventory sub-routes
# ---------------------------------------------------------------------------
@router.get("/{product_id}/inventory", response_model=list[ProductInventory])
def list_product_inventory(
product_id: UUID, session: Session = Depends(get_session)
):
get_or_404(session, Product, product_id)
stmt = select(ProductInventory).where(ProductInventory.product_id == product_id)
return session.exec(stmt).all()
@router.post(
"/{product_id}/inventory", response_model=ProductInventory, status_code=201
)
def create_product_inventory(
product_id: UUID,
data: InventoryCreate,
session: Session = Depends(get_session),
):
get_or_404(session, Product, product_id)
entry = ProductInventory(
id=uuid4(),
product_id=product_id,
**data.model_dump(),
)
session.add(entry)
session.commit()
session.refresh(entry)
return entry

View file

@ -0,0 +1,216 @@
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 GroomingSchedule, Routine, RoutineStep
from innercontext.models.enums import GroomingAction, PartOfDay
router = APIRouter()
# ---------------------------------------------------------------------------
# Schemas
# ---------------------------------------------------------------------------
class RoutineCreate(SQLModel):
routine_date: date
part_of_day: PartOfDay
notes: Optional[str] = None
class RoutineUpdate(SQLModel):
routine_date: Optional[date] = None
part_of_day: Optional[PartOfDay] = None
notes: Optional[str] = None
class RoutineStepCreate(SQLModel):
product_id: Optional[UUID] = None
order_index: int
action_type: Optional[GroomingAction] = None
action_notes: Optional[str] = None
dose: Optional[str] = None
region: Optional[str] = None
class RoutineStepUpdate(SQLModel):
product_id: Optional[UUID] = None
order_index: Optional[int] = None
action_type: Optional[GroomingAction] = None
action_notes: Optional[str] = None
dose: Optional[str] = None
region: Optional[str] = None
class GroomingScheduleCreate(SQLModel):
day_of_week: int
action: GroomingAction
notes: Optional[str] = None
class GroomingScheduleUpdate(SQLModel):
day_of_week: Optional[int] = None
action: Optional[GroomingAction] = 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
# ---------------------------------------------------------------------------
# Routine routes
# ---------------------------------------------------------------------------
@router.get("/", response_model=list[Routine])
def list_routines(
from_date: Optional[date] = None,
to_date: Optional[date] = None,
part_of_day: Optional[PartOfDay] = None,
session: Session = Depends(get_session),
):
stmt = select(Routine)
if from_date is not None:
stmt = stmt.where(Routine.routine_date >= from_date)
if to_date is not None:
stmt = stmt.where(Routine.routine_date <= to_date)
if part_of_day is not None:
stmt = stmt.where(Routine.part_of_day == part_of_day)
return session.exec(stmt).all()
@router.post("/", response_model=Routine, status_code=201)
def create_routine(data: RoutineCreate, session: Session = Depends(get_session)):
routine = Routine(id=uuid4(), **data.model_dump())
session.add(routine)
session.commit()
session.refresh(routine)
return routine
# Grooming-schedule GET must appear before /{routine_id} to avoid being shadowed
@router.get("/grooming-schedule", response_model=list[GroomingSchedule])
def list_grooming_schedule(session: Session = Depends(get_session)):
return session.exec(select(GroomingSchedule)).all()
@router.get("/{routine_id}", response_model=Routine)
def get_routine(routine_id: UUID, session: Session = Depends(get_session)):
return get_or_404(session, Routine, routine_id)
@router.patch("/{routine_id}", response_model=Routine)
def update_routine(
routine_id: UUID,
data: RoutineUpdate,
session: Session = Depends(get_session),
):
routine = get_or_404(session, Routine, routine_id)
for key, value in data.model_dump(exclude_unset=True).items():
setattr(routine, key, value)
session.add(routine)
session.commit()
session.refresh(routine)
return routine
@router.delete("/{routine_id}", status_code=204)
def delete_routine(routine_id: UUID, session: Session = Depends(get_session)):
routine = get_or_404(session, Routine, routine_id)
session.delete(routine)
session.commit()
# ---------------------------------------------------------------------------
# RoutineStep sub-routes
# ---------------------------------------------------------------------------
@router.post("/{routine_id}/steps", response_model=RoutineStep, status_code=201)
def add_step(
routine_id: UUID,
data: RoutineStepCreate,
session: Session = Depends(get_session),
):
get_or_404(session, Routine, routine_id)
step = RoutineStep(id=uuid4(), routine_id=routine_id, **data.model_dump())
session.add(step)
session.commit()
session.refresh(step)
return step
@router.patch("/steps/{step_id}", response_model=RoutineStep)
def update_step(
step_id: UUID,
data: RoutineStepUpdate,
session: Session = Depends(get_session),
):
step = get_or_404(session, RoutineStep, step_id)
for key, value in data.model_dump(exclude_unset=True).items():
setattr(step, key, value)
session.add(step)
session.commit()
session.refresh(step)
return step
@router.delete("/steps/{step_id}", status_code=204)
def delete_step(step_id: UUID, session: Session = Depends(get_session)):
step = get_or_404(session, RoutineStep, step_id)
session.delete(step)
session.commit()
# ---------------------------------------------------------------------------
# GroomingSchedule routes
# ---------------------------------------------------------------------------
@router.post("/grooming-schedule", response_model=GroomingSchedule, status_code=201)
def create_grooming_schedule(
data: GroomingScheduleCreate, session: Session = Depends(get_session)
):
entry = GroomingSchedule(id=uuid4(), **data.model_dump())
session.add(entry)
session.commit()
session.refresh(entry)
return entry
@router.patch("/grooming-schedule/{entry_id}", response_model=GroomingSchedule)
def update_grooming_schedule(
entry_id: UUID,
data: GroomingScheduleUpdate,
session: Session = Depends(get_session),
):
entry = get_or_404(session, GroomingSchedule, entry_id)
for key, value in data.model_dump(exclude_unset=True).items():
setattr(entry, key, value)
session.add(entry)
session.commit()
session.refresh(entry)
return entry
@router.delete("/grooming-schedule/{entry_id}", status_code=204)
def delete_grooming_schedule(
entry_id: UUID, session: Session = Depends(get_session)
):
entry = get_or_404(session, GroomingSchedule, entry_id)
session.delete(entry)
session.commit()

View file

@ -0,0 +1,133 @@
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()

View file

@ -0,0 +1,77 @@
from .domain import Domain
from .enums import (
AbsorptionSpeed,
BarrierState,
DayTime,
EvidenceLevel,
GroomingAction,
IngredientFunction,
InteractionScope,
MedicationKind,
OverallSkinState,
PartOfDay,
PriceTier,
ProductCategory,
ResultFlag,
RoutineRole,
SkinConcern,
SkinTrend,
SkinType,
StrengthLevel,
TextureType,
UsageFrequency,
)
from .health import LabResult, MedicationEntry, MedicationUsage
from .product import (
ActiveIngredient,
Product,
ProductContext,
ProductEffectProfile,
ProductInteraction,
ProductInventory,
)
from .routine import GroomingSchedule, Routine, RoutineStep
from .skincare import SkinConditionSnapshot
__all__ = [
# domain
"Domain",
# enums
"AbsorptionSpeed",
"BarrierState",
"DayTime",
"EvidenceLevel",
"GroomingAction",
"IngredientFunction",
"InteractionScope",
"MedicationKind",
"OverallSkinState",
"PartOfDay",
"PriceTier",
"ProductCategory",
"ResultFlag",
"RoutineRole",
"SkinConcern",
"SkinTrend",
"SkinType",
"StrengthLevel",
"TextureType",
"UsageFrequency",
# health
"LabResult",
"MedicationEntry",
"MedicationUsage",
# product
"ActiveIngredient",
"Product",
"ProductContext",
"ProductEffectProfile",
"ProductInteraction",
"ProductInventory",
# routine
"GroomingSchedule",
"Routine",
"RoutineStep",
# skincare
"SkinConditionSnapshot",
]

View file

@ -0,0 +1,5 @@
from datetime import datetime, timezone
def utc_now() -> datetime:
return datetime.now(timezone.utc)

View file

@ -0,0 +1,6 @@
from enum import Enum
class Domain(str, Enum):
HEALTH = "health"
SKINCARE = "skincare"

View file

@ -0,0 +1,200 @@
from enum import Enum
# ---------------------------------------------------------------------------
# Shared / Product
# ---------------------------------------------------------------------------
class StrengthLevel(int, Enum):
LOW = 1
MEDIUM = 2
HIGH = 3
class AbsorptionSpeed(str, Enum):
VERY_FAST = "very_fast"
FAST = "fast"
MODERATE = "moderate"
SLOW = "slow"
VERY_SLOW = "very_slow"
class UsageFrequency(str, Enum):
DAILY = "daily"
TWICE_DAILY = "twice_daily"
EVERY_OTHER_DAY = "every_other_day"
TWICE_WEEKLY = "twice_weekly"
THREE_TIMES_WEEKLY = "three_times_weekly"
WEEKLY = "weekly"
AS_NEEDED = "as_needed"
class ProductCategory(str, Enum):
CLEANSER = "cleanser"
TONER = "toner"
ESSENCE = "essence"
SERUM = "serum"
MOISTURIZER = "moisturizer"
SPF = "spf"
MASK = "mask"
EXFOLIANT = "exfoliant"
HAIR_TREATMENT = "hair_treatment"
TOOL = "tool"
SPOT_TREATMENT = "spot_treatment"
OIL = "oil"
class DayTime(str, Enum):
AM = "am"
PM = "pm"
BOTH = "both"
class SkinType(str, Enum):
DRY = "dry"
OILY = "oily"
COMBINATION = "combination"
SENSITIVE = "sensitive"
NORMAL = "normal"
ACNE_PRONE = "acne_prone"
class SkinConcern(str, Enum):
ACNE = "acne"
ROSACEA = "rosacea"
HYPERPIGMENTATION = "hyperpigmentation"
AGING = "aging"
DEHYDRATION = "dehydration"
REDNESS = "redness"
DAMAGED_BARRIER = "damaged_barrier"
PORE_VISIBILITY = "pore_visibility"
UNEVEN_TEXTURE = "uneven_texture"
HAIR_GROWTH = "hair_growth"
SEBUM_EXCESS = "sebum_excess"
class IngredientFunction(str, Enum):
HUMECTANT = "humectant"
EMOLLIENT = "emollient"
OCCLUSIVE = "occlusive"
EXFOLIANT_AHA = "exfoliant_aha"
EXFOLIANT_BHA = "exfoliant_bha"
EXFOLIANT_PHA = "exfoliant_pha"
RETINOID = "retinoid"
ANTIOXIDANT = "antioxidant"
SOOTHING = "soothing"
BARRIER_SUPPORT = "barrier_support"
BRIGHTENING = "brightening"
ANTI_ACNE = "anti_acne"
CERAMIDE = "ceramide"
NIACINAMIDE = "niacinamide"
SUNSCREEN = "sunscreen"
PEPTIDE = "peptide"
HAIR_GROWTH_STIMULANT = "hair_growth_stimulant"
PREBIOTIC = "prebiotic"
VITAMIN_C = "vitamin_c"
class TextureType(str, Enum):
WATERY = "watery"
GEL = "gel"
EMULSION = "emulsion"
CREAM = "cream"
OIL = "oil"
BALM = "balm"
FOAM = "foam"
FLUID = "fluid"
class RoutineRole(str, Enum):
CLEANSE = "cleanse"
PREPARE = "prepare"
TREATMENT_ACTIVE = "treatment_active"
TREATMENT_SUPPORT = "treatment_support"
SEAL = "seal"
PROTECT = "protect"
HAIR_TREATMENT = "hair_treatment"
class PriceTier(str, Enum):
BUDGET = "budget"
MID = "mid"
PREMIUM = "premium"
LUXURY = "luxury"
class EvidenceLevel(str, Enum):
LOW = "low"
MIXED = "mixed"
MODERATE = "moderate"
HIGH = "high"
class InteractionScope(str, Enum):
SAME_STEP = "same_step"
SAME_DAY = "same_day"
SAME_PERIOD = "same_period"
# ---------------------------------------------------------------------------
# Health
# ---------------------------------------------------------------------------
class ResultFlag(str, Enum):
NORMAL = "N"
ABNORMAL = "ABN"
POSITIVE = "POS"
NEGATIVE = "NEG"
LOW = "L"
HIGH = "H"
class MedicationKind(str, Enum):
PRESCRIPTION = "prescription"
OTC = "otc"
SUPPLEMENT = "supplement"
HERBAL = "herbal"
OTHER = "other"
# ---------------------------------------------------------------------------
# Routine
# ---------------------------------------------------------------------------
class PartOfDay(str, Enum):
AM = "am"
PM = "pm"
class GroomingAction(str, Enum):
SHAVING_RAZOR = "shaving_razor"
SHAVING_ONEBLADE = "shaving_oneblade"
DERMAROLLING = "dermarolling"
# ---------------------------------------------------------------------------
# Skincare
# ---------------------------------------------------------------------------
class OverallSkinState(str, Enum):
EXCELLENT = "excellent"
GOOD = "good"
FAIR = "fair"
POOR = "poor"
class SkinTrend(str, Enum):
IMPROVING = "improving"
STABLE = "stable"
WORSENING = "worsening"
FLUCTUATING = "fluctuating"
class BarrierState(str, Enum):
INTACT = "intact"
MILDLY_COMPROMISED = "mildly_compromised"
COMPROMISED = "compromised"

View file

@ -0,0 +1,113 @@
from datetime import datetime
from typing import ClassVar
from uuid import UUID, uuid4
from sqlalchemy import Column, DateTime
from sqlmodel import Field, Relationship, SQLModel
from .base import utc_now
from .domain import Domain
from .enums import MedicationKind, ResultFlag
class MedicationEntry(SQLModel, table=True):
__tablename__ = "medication_entries"
__domains__: ClassVar[frozenset[Domain]] = frozenset({Domain.HEALTH})
record_id: UUID = Field(default_factory=uuid4, primary_key=True)
kind: MedicationKind = Field(index=True)
product_name: str = Field(index=True)
active_substance: str | None = Field(default=None, index=True)
formulation: str | None = Field(default=None)
route: str | None = Field(default=None)
source_file: str | None = Field(default=None)
notes: str | None = Field(default=None)
created_at: datetime = Field(default_factory=utc_now, nullable=False)
updated_at: datetime = Field(
default_factory=utc_now,
sa_column=Column(
DateTime(timezone=True),
default=utc_now,
onupdate=utc_now,
nullable=False,
),
)
usage_history: list["MedicationUsage"] = Relationship(back_populates="medication")
class MedicationUsage(SQLModel, table=True):
__tablename__ = "medication_usages"
__domains__: ClassVar[frozenset[Domain]] = frozenset({Domain.HEALTH})
record_id: UUID = Field(default_factory=uuid4, primary_key=True)
medication_record_id: UUID = Field(
foreign_key="medication_entries.record_id", index=True
)
dose_value: float | None = Field(default=None, ge=0)
dose_unit: str | None = Field(default=None)
frequency: str | None = Field(default=None)
schedule_text: str | None = Field(default=None)
as_needed: bool = Field(default=False, index=True)
valid_from: datetime = Field(index=True)
valid_to: datetime | None = Field(default=None, index=True)
source_file: str | None = Field(default=None)
notes: str | None = Field(default=None)
created_at: datetime = Field(default_factory=utc_now, nullable=False)
updated_at: datetime = Field(
default_factory=utc_now,
sa_column=Column(
DateTime(timezone=True),
default=utc_now,
onupdate=utc_now,
nullable=False,
),
)
medication: MedicationEntry = Relationship(back_populates="usage_history")
class LabResult(SQLModel, table=True):
__tablename__ = "lab_results"
__domains__: ClassVar[frozenset[Domain]] = frozenset({Domain.HEALTH})
record_id: UUID = Field(default_factory=uuid4, primary_key=True)
collected_at: datetime = Field(index=True)
test_code: str = Field(index=True, regex=r"^\d+-\d$")
test_name_original: str | None = Field(default=None)
test_name_loinc: str | None = Field(default=None)
value_num: float | None = Field(default=None)
value_text: str | None = Field(default=None)
value_bool: bool | None = Field(default=None)
unit_original: str | None = Field(default=None)
unit_ucum: str | None = Field(default=None)
ref_low: float | None = Field(default=None)
ref_high: float | None = Field(default=None)
ref_text: str | None = Field(default=None)
flag: ResultFlag | None = Field(default=None, index=True)
lab: str | None = Field(default=None, index=True)
source_file: str | None = Field(default=None)
notes: str | None = Field(default=None)
created_at: datetime = Field(default_factory=utc_now, nullable=False)
updated_at: datetime = Field(
default_factory=utc_now,
sa_column=Column(
DateTime(timezone=True),
default=utc_now,
onupdate=utc_now,
nullable=False,
),
)

View file

@ -0,0 +1,377 @@
from datetime import date, datetime
from typing import ClassVar, Optional
from uuid import UUID, uuid4
from pydantic import model_validator
from sqlalchemy import JSON, Column, DateTime
from sqlmodel import Field, Relationship, SQLModel
from .base import utc_now
from .domain import Domain
from .enums import (
AbsorptionSpeed,
DayTime,
EvidenceLevel,
IngredientFunction,
InteractionScope,
PriceTier,
ProductCategory,
RoutineRole,
SkinConcern,
SkinType,
StrengthLevel,
TextureType,
UsageFrequency,
)
# ---------------------------------------------------------------------------
# Value objects
# ---------------------------------------------------------------------------
class ProductEffectProfile(SQLModel):
hydration_immediate: int = Field(default=0, ge=0, le=5)
hydration_long_term: int = Field(default=0, ge=0, le=5)
barrier_repair_strength: int = Field(default=0, ge=0, le=5)
soothing_strength: int = Field(default=0, ge=0, le=5)
exfoliation_strength: int = Field(default=0, ge=0, le=5)
retinoid_strength: int = Field(default=0, ge=0, le=5)
irritation_risk: int = Field(default=0, ge=0, le=5)
comedogenic_risk: int = Field(default=0, ge=0, le=5)
barrier_disruption_risk: int = Field(default=0, ge=0, le=5)
dryness_risk: int = Field(default=0, ge=0, le=5)
brightening_strength: int = Field(default=0, ge=0, le=5)
anti_acne_strength: int = Field(default=0, ge=0, le=5)
anti_aging_strength: int = Field(default=0, ge=0, le=5)
class ActiveIngredient(SQLModel):
name: str
percent: float | None = Field(default=None, ge=0, le=100)
functions: list[IngredientFunction] = Field(default_factory=list)
strength_level: StrengthLevel | None = None
irritation_potential: StrengthLevel | None = None
cumulative_with: list[IngredientFunction] | None = None
class ProductInteraction(SQLModel):
target: str
scope: InteractionScope
reason: str | None = None
class ProductContext(SQLModel):
safe_after_shaving: bool | None = None
safe_after_acids: bool | None = None
safe_after_retinoids: bool | None = None
safe_with_compromised_barrier: bool | None = None
low_uv_only: bool | None = None
# ---------------------------------------------------------------------------
# Helper
# ---------------------------------------------------------------------------
def _ev(v: object) -> str:
"""Return enum value or string as-is (handles both DB-loaded dicts and Python enums)."""
return v.value if hasattr(v, "value") else str(v) # type: ignore[union-attr]
# ---------------------------------------------------------------------------
# Table models
# ---------------------------------------------------------------------------
class Product(SQLModel, table=True):
__tablename__ = "products"
__domains__: ClassVar[frozenset[Domain]] = frozenset({Domain.SKINCARE})
id: UUID = Field(default_factory=uuid4, primary_key=True)
name: str
brand: str
line_name: str | None = Field(default=None, max_length=128)
sku: str | None = Field(default=None, max_length=64)
url: str | None = Field(default=None, max_length=512)
barcode: str | None = Field(default=None, max_length=64)
category: ProductCategory
routine_role: RoutineRole
recommended_time: DayTime
texture: TextureType | None = None
absorption_speed: AbsorptionSpeed | None = None
leave_on: bool
price_tier: PriceTier | None = Field(default=None, index=True)
size_ml: float | None = Field(default=None, gt=0)
pao_months: int | None = Field(default=None, ge=1, le=60)
inci: list[str] = Field(
default_factory=list, sa_column=Column(JSON, nullable=False)
)
actives: list[ActiveIngredient] | None = Field(
default=None, sa_column=Column(JSON, nullable=True)
)
recommended_for: list[SkinType] = Field(
default_factory=list, sa_column=Column(JSON, nullable=False)
)
recommended_frequency: UsageFrequency | None = None
targets: list[SkinConcern] = Field(
default_factory=list, sa_column=Column(JSON, nullable=False)
)
contraindications: list[str] = Field(
default_factory=list, sa_column=Column(JSON, nullable=False)
)
usage_notes: str | None = None
evidence_level: EvidenceLevel | None = Field(default=None, index=True)
claims: list[str] = Field(
default_factory=list, sa_column=Column(JSON, nullable=False)
)
fragrance_free: bool | None = None
essential_oils_free: bool | None = None
alcohol_denat_free: bool | None = None
pregnancy_safe: bool | None = None
product_effect_profile: ProductEffectProfile = Field(
default_factory=ProductEffectProfile,
sa_column=Column(JSON, nullable=False),
)
ph_min: float | None = Field(default=None, ge=0, le=14)
ph_max: float | None = Field(default=None, ge=0, le=14)
incompatible_with: list[ProductInteraction] | None = Field(
default=None, sa_column=Column(JSON, nullable=True)
)
synergizes_with: list[str] | None = Field(
default=None, sa_column=Column(JSON, nullable=True)
)
context_rules: ProductContext | None = Field(
default=None, sa_column=Column(JSON, nullable=True)
)
min_interval_hours: int | None = Field(default=None, ge=0)
max_frequency_per_week: int | None = Field(default=None, ge=1, le=14)
is_medication: bool = Field(default=False)
is_tool: bool = Field(default=False)
needle_length_mm: float | None = Field(default=None, gt=0)
personal_rating: int | None = Field(default=None, ge=1, le=10)
personal_tolerance_notes: str | None = None
personal_repurchase_intent: bool | None = None
created_at: datetime = Field(default_factory=utc_now, nullable=False)
updated_at: datetime = Field(
default_factory=utc_now,
sa_column=Column(
DateTime(timezone=True),
default=utc_now,
onupdate=utc_now,
nullable=False,
),
)
inventory: list["ProductInventory"] = Relationship(back_populates="product")
@model_validator(mode="after")
def validate_business_rules(self) -> "Product":
if (
self.ph_min is not None
and self.ph_max is not None
and self.ph_min > self.ph_max
):
raise ValueError("ph_min must be <= ph_max")
if self.category == ProductCategory.SPF and self.recommended_time == DayTime.PM:
raise ValueError("SPF cannot be recommended only for PM")
if self.category == ProductCategory.SPF and not self.leave_on:
raise ValueError("SPF products must be leave-on")
if self.is_medication and not self.usage_notes:
raise ValueError("Medication products must have usage_notes")
return self
def to_llm_context(self) -> dict:
ctx: dict = {
"id": str(self.id),
"name": self.name,
"brand": self.brand,
"category": _ev(self.category),
"routine_role": _ev(self.routine_role),
"recommended_time": _ev(self.recommended_time),
"leave_on": self.leave_on,
}
for field in ("line_name", "sku", "url", "barcode"):
val = getattr(self, field)
if val is not None:
ctx[field] = val
if self.texture is not None:
ctx["texture"] = _ev(self.texture)
if self.absorption_speed is not None:
ctx["absorption_speed"] = _ev(self.absorption_speed)
if self.price_tier is not None:
ctx["price_tier"] = _ev(self.price_tier)
if self.size_ml is not None:
ctx["size_ml"] = self.size_ml
if self.pao_months is not None:
ctx["pao_months"] = self.pao_months
if self.recommended_frequency is not None:
ctx["recommended_frequency"] = _ev(self.recommended_frequency)
if self.evidence_level is not None:
ctx["evidence_level"] = _ev(self.evidence_level)
if self.inci:
ctx["inci"] = self.inci
if self.recommended_for:
ctx["recommended_for"] = [_ev(s) for s in self.recommended_for]
if self.targets:
ctx["targets"] = [_ev(s) for s in self.targets]
if self.contraindications:
ctx["contraindications"] = self.contraindications
if self.claims:
ctx["claims"] = self.claims
if self.actives:
actives_ctx = []
for a in self.actives:
if isinstance(a, dict):
actives_ctx.append(a)
else:
a_dict: dict = {"name": a.name}
if a.percent is not None:
a_dict["percent"] = a.percent
if a.functions:
a_dict["functions"] = [_ev(f) for f in a.functions]
if a.strength_level is not None:
a_dict["strength_level"] = a.strength_level.name.lower()
if a.cumulative_with:
a_dict["cumulative_with"] = [_ev(f) for f in a.cumulative_with]
actives_ctx.append(a_dict)
ctx["actives"] = actives_ctx
if self.ph_min is not None or self.ph_max is not None:
if self.ph_min == self.ph_max and self.ph_min is not None:
ctx["ph"] = self.ph_min
elif self.ph_min is not None and self.ph_max is not None:
ctx["ph_range"] = f"{self.ph_min}{self.ph_max}"
elif self.ph_min is not None:
ctx["ph_min"] = self.ph_min
else:
ctx["ph_max"] = self.ph_max
ep = self.product_effect_profile
if ep is not None:
if isinstance(ep, dict):
nonzero = {k: v for k, v in ep.items() if v}
else:
nonzero = {k: v for k, v in ep.model_dump().items() if v}
if nonzero:
ctx["effect_profile"] = nonzero
if self.incompatible_with:
parts = []
for inc in self.incompatible_with:
if isinstance(inc, dict):
scope = inc.get("scope", "")
target = inc.get("target", "")
reason = inc.get("reason")
else:
scope = _ev(inc.scope)
target = inc.target
reason = inc.reason
msg = f"avoid {target} ({scope})"
if reason:
msg += f": {reason}"
parts.append(msg)
ctx["incompatible_with"] = parts
if self.synergizes_with:
ctx["synergizes_with"] = self.synergizes_with
if self.context_rules is not None:
cr = self.context_rules
if isinstance(cr, dict):
rules = {k: v for k, v in cr.items() if v is not None}
else:
rules = {k: v for k, v in cr.model_dump().items() if v is not None}
if rules:
ctx["context_rules"] = rules
if self.min_interval_hours is not None:
ctx["min_interval_hours"] = self.min_interval_hours
if self.max_frequency_per_week is not None:
ctx["max_frequency_per_week"] = self.max_frequency_per_week
safety = {}
for flag in ("fragrance_free", "essential_oils_free", "alcohol_denat_free", "pregnancy_safe"):
val = getattr(self, flag)
if val is not None:
safety[flag] = val
if safety:
ctx["safety"] = safety
if self.is_medication:
ctx["is_medication"] = True
if self.is_tool:
ctx["is_tool"] = True
if self.needle_length_mm is not None:
ctx["needle_length_mm"] = self.needle_length_mm
if self.usage_notes:
ctx["usage_notes"] = self.usage_notes
if self.personal_rating is not None:
ctx["personal_rating"] = self.personal_rating
if self.personal_tolerance_notes:
ctx["personal_tolerance_notes"] = self.personal_tolerance_notes
if self.personal_repurchase_intent is not None:
ctx["personal_repurchase_intent"] = self.personal_repurchase_intent
try:
opened_items = [
inv for inv in (self.inventory or []) if inv.is_opened and inv.opened_at
]
if opened_items:
most_recent = max(opened_items, key=lambda x: x.opened_at)
ctx["days_since_opened"] = (date.today() - most_recent.opened_at).days
except Exception:
pass
return ctx
class ProductInventory(SQLModel, table=True):
__tablename__ = "product_inventory"
__domains__: ClassVar[frozenset[Domain]] = frozenset({Domain.SKINCARE})
id: UUID = Field(default_factory=uuid4, primary_key=True)
product_id: UUID = Field(foreign_key="products.id", index=True)
is_opened: bool = Field(default=False)
opened_at: date | None = Field(default=None)
finished_at: date | None = Field(default=None)
expiry_date: date | None = Field(default=None)
current_weight_g: float | None = Field(default=None, gt=0)
notes: str | None = None
created_at: datetime = Field(default_factory=utc_now, nullable=False)
product: Optional["Product"] = Relationship(back_populates="inventory")

View file

@ -0,0 +1,70 @@
from datetime import date, datetime
from typing import TYPE_CHECKING, ClassVar, List, Optional
from uuid import UUID, uuid4
from sqlalchemy import Column, DateTime, UniqueConstraint
from sqlmodel import Field, Relationship, SQLModel
from .base import utc_now
from .domain import Domain
from .enums import GroomingAction, PartOfDay
if TYPE_CHECKING:
from .product import Product
class Routine(SQLModel, table=True):
__tablename__ = "routines"
__domains__: ClassVar[frozenset[Domain]] = frozenset({Domain.SKINCARE})
__table_args__ = (
UniqueConstraint(
"routine_date", "part_of_day", name="uq_routine_date_part_of_day"
),
)
id: UUID = Field(default_factory=uuid4, primary_key=True)
routine_date: date = Field(index=True)
part_of_day: PartOfDay = Field(index=True)
notes: str | None = Field(default=None)
created_at: datetime = Field(default_factory=utc_now, nullable=False)
updated_at: datetime = Field(
default_factory=utc_now,
sa_column=Column(
DateTime(timezone=True),
default=utc_now,
onupdate=utc_now,
nullable=False,
),
)
steps: List["RoutineStep"] = Relationship(back_populates="routine")
class GroomingSchedule(SQLModel, table=True):
__tablename__ = "grooming_schedule"
__domains__: ClassVar[frozenset[Domain]] = frozenset({Domain.SKINCARE})
id: UUID = Field(default_factory=uuid4, primary_key=True)
day_of_week: int = Field(ge=0, le=6, index=True) # 0 = poniedziałek, 6 = niedziela
action: GroomingAction
notes: str | None = Field(default=None)
class RoutineStep(SQLModel, table=True):
__tablename__ = "routine_steps"
__domains__: ClassVar[frozenset[Domain]] = frozenset({Domain.SKINCARE})
id: UUID = Field(default_factory=uuid4, primary_key=True)
routine_id: UUID = Field(foreign_key="routines.id", index=True)
product_id: UUID | None = Field(default=None, foreign_key="products.id", index=True)
order_index: int = Field(ge=0)
action_type: GroomingAction | None = Field(default=None)
action_notes: str | None = Field(default=None)
dose: str | None = Field(default=None)
region: str | None = Field(default=None)
routine: Routine = Relationship(back_populates="steps")
product: Optional["Product"] = Relationship()

View file

@ -0,0 +1,54 @@
from __future__ import annotations
from datetime import date, datetime
from typing import ClassVar
from uuid import UUID, uuid4
from sqlalchemy import JSON, Column, UniqueConstraint
from sqlmodel import Field, SQLModel
from .base import utc_now
from .domain import Domain
from .enums import BarrierState, OverallSkinState, SkinConcern, SkinTrend, SkinType
class SkinConditionSnapshot(SQLModel, table=True):
"""
Tygodniowy snapshot kondycji skóry wypełniany przez LLM na podstawie zdjęć
i kontekstu rutyny. Wszystkie metryki numeryczne w skali 15.
"""
__tablename__ = "skin_condition_snapshots"
__domains__: ClassVar[frozenset[Domain]] = frozenset({Domain.SKINCARE})
__table_args__ = (UniqueConstraint("snapshot_date", name="uq_skin_snapshot_date"),)
id: UUID = Field(default_factory=uuid4, primary_key=True)
snapshot_date: date = Field(index=True)
overall_state: OverallSkinState | None = None
trend: SkinTrend | None = None
skin_type: SkinType | None = None
# Metryki wizualne (1 = minimalne, 5 = maksymalne nasilenie)
hydration_level: int | None = Field(default=None, ge=1, le=5)
sebum_tzone: int | None = Field(default=None, ge=1, le=5)
sebum_cheeks: int | None = Field(default=None, ge=1, le=5)
sensitivity_level: int | None = Field(default=None, ge=1, le=5)
barrier_state: BarrierState | None = None
# Aktywne troski — podzbiór SkinConcern widoczny na zdjęciach lub wynikający z rutyny
active_concerns: list[SkinConcern] = Field(
default_factory=list, sa_column=Column(JSON, nullable=False)
)
# Wolny tekst — LLM wypełnia na podstawie analizy
risks: list[str] = Field(
default_factory=list, sa_column=Column(JSON, nullable=False)
)
priorities: list[str] = Field(
default_factory=list, sa_column=Column(JSON, nullable=False)
)
notes: str | None = None
created_at: datetime = Field(default_factory=utc_now, nullable=False)