Craig
250 lines
9.1 KiB
Python
250 lines
9.1 KiB
Python
"""Foods service layer — business logic + DB access (spec §8.1 rules 2–3, 7).
|
||
|
||
ORM objects never leave this module; functions return Pydantic schemas.
|
||
Handlers stay thin (~15 lines) by calling into these functions.
|
||
|
||
Meal foods are returned as MealRead (with components + computed nutrition)
|
||
via get_meal(). get_food() returns the appropriate type based on is_meal.
|
||
"""
|
||
|
||
from datetime import datetime, timezone
|
||
|
||
from sqlalchemy import desc, func, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from models import DailyLogEntry, Food
|
||
from schemas import FoodCreate, FoodRead, FoodUpdate, MealRead
|
||
from services.meals import _build_meal_read, load_meal_with_components
|
||
|
||
|
||
# ── 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 | MealRead | None:
|
||
"""Get a single food by id. INCLUDES soft-deleted foods (historical log
|
||
rendering depends on this — spec §2.1).
|
||
|
||
Returns MealRead (with components + computed nutrition) for meal foods,
|
||
plain FoodRead for non-meals. Returns None for unknown id.
|
||
"""
|
||
food = load_meal_with_components(db, food_id)
|
||
if food is None:
|
||
# Not a meal — try as a plain food
|
||
food = db.get(Food, food_id)
|
||
if food is None:
|
||
return None
|
||
return FoodRead.model_validate(food)
|
||
# Meal food: return MealRead with components
|
||
return _build_meal_read(food)
|
||
|
||
|
||
def get_meal(db: Session, meal_id: int) -> MealRead | None:
|
||
"""Get a meal food with components and computed nutrition.
|
||
|
||
Returns None if the food doesn't exist or is not a meal."""
|
||
meal = load_meal_with_components(db, meal_id)
|
||
if meal is None:
|
||
return None
|
||
return _build_meal_read(meal)
|
||
|
||
|
||
def create_food(db: Session, data: FoodCreate) -> FoodRead:
|
||
"""Create a food. Barcode uniqueness on LIVE foods → 409.
|
||
|
||
Restore-on-rescan (§3.1): when the barcode matches a SOFT-DELETED food
|
||
(deleted_at IS NOT NULL), clear deleted_at and UPDATE that existing row
|
||
with the new data instead of inserting a duplicate. All in one transaction.
|
||
|
||
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:
|
||
if existing.deleted_at is not None:
|
||
# ── Restore-on-rescan: update the soft-deleted row ──
|
||
_apply_create_data(existing, data)
|
||
existing.deleted_at = None
|
||
existing.updated_at = _now()
|
||
db.commit()
|
||
db.refresh(existing)
|
||
return FoodRead.model_validate(existing)
|
||
# Live food with same barcode → conflict
|
||
raise BarcodeConflictError(data.barcode)
|
||
|
||
now = _now()
|
||
food = Food()
|
||
_apply_create_data(food, data)
|
||
food.created_at = now
|
||
food.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)
|
||
|
||
|
||
def restore_food(db: Session, food_id: int) -> FoodRead | None:
|
||
"""Restore a soft-deleted food: clears deleted_at (spec §3.1).
|
||
Idempotent — restoring a live food just returns it. Returns None if not found."""
|
||
food = db.get(Food, food_id)
|
||
if food is None:
|
||
return None
|
||
|
||
if food.deleted_at is not None:
|
||
food.deleted_at = None
|
||
food.updated_at = datetime.now(timezone.utc).replace(tzinfo=None)
|
||
db.commit()
|
||
db.refresh(food)
|
||
return FoodRead.model_validate(food)
|
||
|
||
|
||
# ── Recent foods (TICKET-006) ────────────────────────────────────────────────
|
||
|
||
|
||
def get_recent_foods(db: Session, limit: int = 10) -> list[FoodRead]:
|
||
"""Return foods ordered by most recent appearance in daily_log.
|
||
|
||
Deduplicated: each food appears at most once.
|
||
Excludes soft-deleted foods (§8.1 rule 7).
|
||
|
||
Ordered by the daily_log entry's created_at (the moment the food was
|
||
logged), NOT by the food's own created_at and NOT by the log ``date``
|
||
field. This means a food you just backfilled to an old date still
|
||
appears as "recent" because your log *action* was recent — which is
|
||
the intended UX.
|
||
|
||
If no foods have ever been logged, returns an empty list.
|
||
"""
|
||
last_logged = func.max(DailyLogEntry.created_at).label("last_logged")
|
||
stmt = (
|
||
select(Food, last_logged)
|
||
.join(DailyLogEntry, DailyLogEntry.food_id == Food.id)
|
||
.where(_not_deleted())
|
||
.group_by(Food.id)
|
||
.order_by(desc("last_logged"))
|
||
.limit(limit)
|
||
)
|
||
rows = db.execute(stmt).all()
|
||
return [FoodRead.model_validate(row[0]) for row in rows]
|
||
|
||
|
||
# ── Internal helpers ─────────────────────────────────────────────────────────
|
||
|
||
|
||
def _now() -> datetime:
|
||
return datetime.now(timezone.utc).replace(tzinfo=None)
|
||
|
||
|
||
def _apply_create_data(food: Food, data: FoodCreate) -> None:
|
||
"""Copy FoodCreate fields onto a Food ORM object (used for both insert
|
||
and restore-on-rescan update paths)."""
|
||
food.name = data.name
|
||
food.brand = data.brand
|
||
food.barcode = data.barcode
|
||
food.source = data.source
|
||
food.is_meal = data.is_meal
|
||
food.unit_type = data.unit_type
|
||
food.calories_per_unit = data.calories_per_unit
|
||
food.protein_per_unit = data.protein_per_unit
|
||
food.carbs_per_unit = data.carbs_per_unit
|
||
food.fat_per_unit = data.fat_per_unit
|
||
food.fiber_per_unit = data.fiber_per_unit
|
||
food.saturated_fat_per_unit = data.saturated_fat_per_unit
|
||
food.sugars_per_unit = data.sugars_per_unit
|
||
food.sodium_per_unit = data.sodium_per_unit
|
||
food.serving_size_g = data.serving_size_g
|
||
food.serving_name = data.serving_name
|
||
food.off_data = data.off_data
|
||
|
||
|
||
# ── 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
|