TICKET-001: Foods CRUD with soft-delete

- POST/GET/PUT/DELETE /api/foods per spec 3.1 (minus restore)
- Service layer (services/foods.py) with shared soft-delete query helper
- FoodCreate extended, FoodUpdate/FoodRead schemas added
- Barcode conflict returns 409; deleted foods hidden from search,
  visible by id and via include_deleted=true
- 29 new tests, full suite green (36 passed)
This commit is contained in:
Craig
2026-07-26 12:56:59 +01:00
parent 3ab08c4c90
commit 3f2c765c84
5 changed files with 589 additions and 13 deletions
+154
View File
@@ -0,0 +1,154 @@
"""Foods service layer — business logic + DB access (spec §8.1 rules 23, 7).
ORM objects never leave this module; functions return Pydantic schemas.
Handlers stay thin (~15 lines) by calling into these functions.
"""
from datetime import datetime, timezone
from sqlalchemy import select
from sqlalchemy.orm import Session
from models import Food
from schemas import FoodCreate, FoodRead, FoodUpdate
# ── Shared query helpers (§8.1 rule 7) ───────────────────────────────────────
def _not_deleted():
"""Shared filter: exclude soft-deleted foods. Use everywhere search/recent
queries filter — never copy-paste per-endpoint."""
return Food.deleted_at.is_(None)
def _base_search_stmt(db: Session, include_deleted: bool = False):
"""Base statement for food queries. Applies soft-delete filter unless
include_deleted is True."""
stmt = select(Food)
if not include_deleted:
stmt = stmt.where(_not_deleted())
return stmt
# ── CRUD ─────────────────────────────────────────────────────────────────────
def list_foods(
db: Session,
q: str | None = None,
barcode: str | None = None,
limit: int = 50,
offset: int = 0,
include_deleted: bool = False,
) -> list[FoodRead]:
"""Search foods. Supports q (matches name AND brand), barcode (exact match),
limit/offset pagination, and include_deleted toggle."""
stmt = _base_search_stmt(db, include_deleted=include_deleted)
if q:
stmt = stmt.where(Food.name.contains(q) | Food.brand.contains(q))
if barcode:
stmt = stmt.where(Food.barcode == barcode)
stmt = stmt.limit(limit).offset(offset)
foods = db.scalars(stmt).all()
return [FoodRead.model_validate(f) for f in foods]
def get_food(db: Session, food_id: int) -> FoodRead | None:
"""Get a single food by id. INCLUDES soft-deleted foods (historical log
rendering depends on this — spec §2.1). Returns None for unknown id."""
food = db.get(Food, food_id)
if food is None:
return None
return FoodRead.model_validate(food)
def create_food(db: Session, data: FoodCreate) -> FoodRead:
"""Create a food. Checks barcode uniqueness on live foods (409).
SQLite treats NULL barcodes as distinct, so multiple barcode-less foods
are fine."""
if data.barcode is not None:
existing = db.scalar(
select(Food).where(Food.barcode == data.barcode)
)
if existing is not None:
raise BarcodeConflictError(data.barcode)
now = datetime.now(timezone.utc).replace(tzinfo=None)
food = Food(
name=data.name,
brand=data.brand,
barcode=data.barcode,
source=data.source,
is_meal=data.is_meal,
unit_type=data.unit_type,
calories_per_unit=data.calories_per_unit,
protein_per_unit=data.protein_per_unit,
carbs_per_unit=data.carbs_per_unit,
fat_per_unit=data.fat_per_unit,
fiber_per_unit=data.fiber_per_unit,
saturated_fat_per_unit=data.saturated_fat_per_unit,
sugars_per_unit=data.sugars_per_unit,
sodium_per_unit=data.sodium_per_unit,
serving_size_g=data.serving_size_g,
serving_name=data.serving_name,
off_data=data.off_data,
created_at=now,
updated_at=now,
)
db.add(food)
db.commit()
db.refresh(food)
return FoodRead.model_validate(food)
def update_food(db: Session, food_id: int, data: FoodUpdate) -> FoodRead | None:
"""Update editable fields on a food. Only supplied (non-None) fields are
changed. Returns None if food not found."""
food = db.get(Food, food_id)
if food is None:
return None
update_data = data.model_dump(exclude_unset=True)
# Check barcode uniqueness if barcode is being changed to a non-None value
if "barcode" in update_data and update_data["barcode"] is not None:
new_bc = update_data["barcode"]
existing = db.scalar(
select(Food).where(Food.barcode == new_bc, Food.id != food_id)
)
if existing is not None:
raise BarcodeConflictError(new_bc)
for field, value in update_data.items():
setattr(food, field, value)
food.updated_at = datetime.now(timezone.utc).replace(tzinfo=None)
db.commit()
db.refresh(food)
return FoodRead.model_validate(food)
def delete_food(db: Session, food_id: int) -> FoodRead | None:
"""Soft-delete a food: sets deleted_at timestamp. Returns None if not found."""
food = db.get(Food, food_id)
if food is None:
return None
food.deleted_at = datetime.now(timezone.utc).replace(tzinfo=None)
food.updated_at = datetime.now(timezone.utc).replace(tzinfo=None)
db.commit()
db.refresh(food)
return FoodRead.model_validate(food)
# ── Errors ───────────────────────────────────────────────────────────────────
class BarcodeConflictError(Exception):
"""Raised when a barcode already exists on another food."""
def __init__(self, barcode: str):
super().__init__(f"Barcode '{barcode}' already exists")
self.barcode = barcode