craig/calcount
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TICKET-007 (backend): meals — from-log, unpack, components, recursive nutrition, cycle detection

Browse Source
This commit is contained in:
Craig
2026-07-26 17:27:15 +01:00
parent 4dd44b08d0
commit a8aed9a84f
9 changed files with 1720 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/foods.py
+28 -5
View File
@@ -2,6 +2,9 @@
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
@@ -10,7 +13,8 @@ from sqlalchemy import desc, func, select
from sqlalchemy.orm import Session
from models import DailyLogEntry, Food
from schemas import FoodCreate, FoodRead, FoodUpdate
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) ───────────────────────────────────────
@@ -56,13 +60,32 @@ def list_foods(
return [FoodRead.model_validate(f) for f in foods]
def get_food(db: Session, food_id: int) -> FoodRead | None:
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 None for unknown id."""
food = db.get(Food, food_id)
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 FoodRead.model_validate(food)
return _build_meal_read(meal)
def create_food(db: Session, data: FoodCreate) -> FoodRead:
backend/services/log.py
+120 -24
View File
@@ -12,15 +12,16 @@ ORM objects never leave this module; functions return Pydantic schemas.
from datetime import date, datetime, timezone
from sqlalchemy import func, select
from sqlalchemy.orm import Session, joinedload
from sqlalchemy.orm import Session, joinedload, selectinload
from models import DailyLogEntry, Food
from models import DailyLogEntry, Food, MealComponent
from schemas import (
DaySummaryNutrition,
DaySummaryResponse,
LogEntryCreate,
LogEntryRead,
LogEntryUpdate,
MealComponentRead,
)
from services import nutrition
from services import targets as targets_svc
@@ -39,20 +40,82 @@ def _load_entry_with_food(db: Session, entry_id: int) -> DailyLogEntry | None:
)
def _build_log_food_dict(food: Food) -> dict:
"""Convert a Food ORM to dict for LogFoodRead, including nested components
when the food is a meal."""
data = {
"id": food.id,
"name": food.name,
"brand": food.brand,
"unit_type": food.unit_type,
"calories_per_unit": food.calories_per_unit,
"serving_size_g": food.serving_size_g,
"serving_name": food.serving_name,
"is_meal": food.is_meal,
"deleted_at": food.deleted_at,
}
if food.is_meal and hasattr(food, "components"):
comps = []
for mc in food.components:
cf = mc.food
comps.append(MealComponentRead(
food_id=mc.food_id,
quantity=mc.quantity,
food={
"id": cf.id,
"name": cf.name,
"brand": cf.brand,
"unit_type": cf.unit_type,
"calories_per_unit": cf.calories_per_unit,
"serving_size_g": cf.serving_size_g,
"serving_name": cf.serving_name,
"is_meal": cf.is_meal,
"deleted_at": cf.deleted_at,
},
))
data["components"] = comps
else:
data["components"] = None
return data
# ── CRUD ─────────────────────────────────────────────────────────────────────
def get_log_entries(db: Session, lookup_date: date) -> list[LogEntryRead]:
"""All entries for a date, ordered by sort_order then id, with embedded
food data eagerly loaded. Soft-deleted foods still render (§2.1)."""
food data eagerly loaded. Meal foods include their nested components
so the frontend can render collapsible rows (§3.3).
Soft-deleted foods still render (§2.1).
"""
stmt = (
select(DailyLogEntry)
.where(DailyLogEntry.date == lookup_date)
.options(joinedload(DailyLogEntry.food))
.options(
joinedload(DailyLogEntry.food)
.selectinload(Food.components)
.joinedload(MealComponent.food)
)
.order_by(DailyLogEntry.sort_order, DailyLogEntry.id)
)
entries = db.scalars(stmt).all()
return [LogEntryRead.model_validate(e) for e in entries]
entries = db.scalars(stmt).unique().all()
result: list[LogEntryRead] = []
for entry in entries:
food_data = _build_log_food_dict(entry.food)
# Compute nutrition for this entry (handles meals recursively)
entry_nut = nutrition.entry_nutrition(entry.food, entry.quantity)
result.append(LogEntryRead.model_validate({
"id": entry.id,
"date": entry.date,
"food_id": entry.food_id,
"quantity": entry.quantity,
"meal_slot": entry.meal_slot,
"sort_order": entry.sort_order,
"food": food_data,
"computed_nutrition": entry_nut,
}))
return result
def create_log_entry(db: Session, data: LogEntryCreate) -> LogEntryRead:
@@ -87,7 +150,19 @@ def create_log_entry(db: Session, data: LogEntryCreate) -> LogEntryRead:
db.commit()
# Re-query with eager-loaded food for the response
return LogEntryRead.model_validate(_load_entry_with_food(db, entry.id))
loaded = _load_entry_with_food(db, entry.id)
food_data = _build_log_food_dict(loaded.food)
entry_nut = nutrition.entry_nutrition(loaded.food, loaded.quantity)
return LogEntryRead.model_validate({
"id": loaded.id,
"date": loaded.date,
"food_id": loaded.food_id,
"quantity": loaded.quantity,
"meal_slot": loaded.meal_slot,
"sort_order": loaded.sort_order,
"food": food_data,
"computed_nutrition": entry_nut,
})
def update_log_entry(
@@ -109,7 +184,29 @@ def update_log_entry(
setattr(entry, field, value)
db.commit()
return LogEntryRead.model_validate(_load_entry_with_food(db, entry_id))
# Re-query with eager-loaded food + components for the response
loaded = db.scalar(
select(DailyLogEntry)
.where(DailyLogEntry.id == entry_id)
.options(
joinedload(DailyLogEntry.food)
.selectinload(Food.components)
.joinedload(MealComponent.food)
)
)
food_data = _build_log_food_dict(loaded.food)
entry_nut = nutrition.entry_nutrition(loaded.food, loaded.quantity)
return LogEntryRead.model_validate({
"id": loaded.id,
"date": loaded.date,
"food_id": loaded.food_id,
"quantity": loaded.quantity,
"meal_slot": loaded.meal_slot,
"sort_order": loaded.sort_order,
"food": food_data,
"computed_nutrition": entry_nut,
})
def delete_log_entry(db: Session, entry_id: int) -> bool:
@@ -141,14 +238,12 @@ class FoodNotAvailableError(Exception):
def get_day_summary(db: Session, lookup_date: date) -> DaySummaryResponse:
"""Compute nutrition totals for a date vs. the applicable target.
1. Loads all log entries for the date with food data eagerly joined.
2. For each entry, scales the food's per-unit nutrition to the logged
quantity via nutrition.entry_nutrition() — weight-type foods get
(qty/100)× scaling, count-type foods get qty× scaling (§2.1).
3. Meal entries (is_meal=True) currently contribute 0 because their
per_unit fields are null per the CHECK constraint.
TODO: TICKET-007 — replace with recursive component summation.
4. Looks up the target covering the date via the half-open interval
1. Loads all log entries for the date with food data and meal components
eagerly joined.
2. For each entry, uses nutrition.entry_nutrition() which now handles
meal foods recursively by summing their component nutrition (§2.2).
Cycle detection prevents infinite loops on inconsistent data.
3. Looks up the target covering the date via the half-open interval
lookup from targets.get_target_for_date(). Returns null if none.
Returns a DaySummaryResponse with summed totals and the applicable
@@ -157,15 +252,16 @@ def get_day_summary(db: Session, lookup_date: date) -> DaySummaryResponse:
entries = db.scalars(
select(DailyLogEntry)
.where(DailyLogEntry.date == lookup_date)
.options(joinedload(DailyLogEntry.food))
).all()
.options(
joinedload(DailyLogEntry.food)
.selectinload(Food.components)
.joinedload(MealComponent.food)
)
).unique().all()
# Sum nutrition across all entries for the date.
# TODO: TICKET-007 — meal entries currently contribute 0 because their
# per_unit fields are null per the CHECK constraint. Real meal nutrition
# will be derived by recursively summing component foods' nutrition.
# When that lands, replace the flat entry_nutrition() call with a
# meal-aware sum function from services/nutrition.py.
# entry_nutrition() now handles meal foods by recursive component
# summation (TICKET-007).
totals = {field: 0.0 for field in nutrition.NUTRITION_FIELDS}
for entry in entries:
entry_nut = nutrition.entry_nutrition(entry.food, entry.quantity)
@@ -178,4 +274,4 @@ def get_day_summary(db: Session, lookup_date: date) -> DaySummaryResponse:
date=lookup_date,
totals=DaySummaryNutrition(**totals),
target=target,
)
)
backend/services/meals.py
+546 -2
View File
@@ -1,7 +1,551 @@
"""Meal business logic: composition, recursion, cycle detection (spec §2.2).
Multi-write operations here own their transactions (spec §8.1 rule 6):
commit once at the end or roll back entirely.
create_meal_from_log, unpack_meal, and update_meal_components each commit once
at the end or roll back entirely — never partial writes.
TODO: from-log, unpack, component replacement with cycle checks.
ORM objects never leave this module; functions return Pydantic schemas
(spec §8.1 rule 3).
"""
from __future__ import annotations
from datetime import date, datetime, timezone
from sqlalchemy import func, select
from sqlalchemy.orm import Session, joinedload, selectinload
from models import DailyLogEntry, Food, MealComponent
from schemas import (
FoodRead,
LogEntryRead,
MealComponentRead,
MealRead,
)
from services import nutrition
# ── Helpers ──────────────────────────────────────────────────────────────────
def _now() -> datetime:
return datetime.now(timezone.utc).replace(tzinfo=None)
def _load_entry_with_food(db: Session, entry_id: int) -> DailyLogEntry | None:
"""Eager-load a single entry with its food relationship."""
return db.scalar(
select(DailyLogEntry)
.where(DailyLogEntry.id == entry_id)
.options(joinedload(DailyLogEntry.food))
)
def _load_entry_with_food_components(db: Session, entry_id: int) -> DailyLogEntry | None:
"""Eager-load a single entry with food and nested meal components."""
return db.scalar(
select(DailyLogEntry)
.where(DailyLogEntry.id == entry_id)
.options(
joinedload(DailyLogEntry.food)
.selectinload(Food.components)
.joinedload(MealComponent.food)
)
)
def _build_log_entry_read(entry: DailyLogEntry) -> LogEntryRead:
"""Convert a DailyLogEntry ORM object to LogEntryRead, populating
nested meal components on the embedded food when is_meal is True."""
data = {
"id": entry.id,
"date": entry.date,
"food_id": entry.food_id,
"quantity": entry.quantity,
"meal_slot": entry.meal_slot,
"sort_order": entry.sort_order,
}
# Build the embedded food
food = entry.food
food_data = {
"id": food.id,
"name": food.name,
"brand": food.brand,
"unit_type": food.unit_type,
"calories_per_unit": food.calories_per_unit,
"serving_size_g": food.serving_size_g,
"serving_name": food.serving_name,
"is_meal": food.is_meal,
"deleted_at": food.deleted_at,
}
# Populate nested components for meal foods
if food.is_meal and hasattr(food, "components"):
comps = []
for mc in food.components:
comp_food = mc.food
comps.append(MealComponentRead(
food_id=mc.food_id,
quantity=mc.quantity,
food={
"id": comp_food.id,
"name": comp_food.name,
"brand": comp_food.brand,
"unit_type": comp_food.unit_type,
"calories_per_unit": comp_food.calories_per_unit,
"serving_size_g": comp_food.serving_size_g,
"serving_name": comp_food.serving_name,
"is_meal": comp_food.is_meal,
"deleted_at": comp_food.deleted_at,
},
))
food_data["components"] = comps
else:
food_data["components"] = None
data["food"] = food_data
# Compute nutrition for this entry (handles meals recursively)
data["computed_nutrition"] = nutrition.entry_nutrition(entry.food, entry.quantity)
return LogEntryRead.model_validate(data)
def _build_meal_read(meal: Food) -> MealRead:
"""Convert a meal Food ORM to MealRead with components & computed nutrition."""
# Base food fields
meal_data = {
"id": meal.id,
"name": meal.name,
"brand": meal.brand,
"barcode": meal.barcode,
"source": meal.source,
"is_meal": meal.is_meal,
"unit_type": meal.unit_type,
"calories_per_unit": meal.calories_per_unit,
"protein_per_unit": meal.protein_per_unit,
"carbs_per_unit": meal.carbs_per_unit,
"fat_per_unit": meal.fat_per_unit,
"fiber_per_unit": meal.fiber_per_unit,
"saturated_fat_per_unit": meal.saturated_fat_per_unit,
"sugars_per_unit": meal.sugars_per_unit,
"sodium_per_unit": meal.sodium_per_unit,
"serving_size_g": meal.serving_size_g,
"serving_name": meal.serving_name,
"deleted_at": meal.deleted_at,
"created_at": meal.created_at,
"updated_at": meal.updated_at,
}
# Build component list
comps = []
if hasattr(meal, "components"):
for mc in meal.components:
comp_food = mc.food
comps.append(MealComponentRead(
food_id=mc.food_id,
quantity=mc.quantity,
food={
"id": comp_food.id,
"name": comp_food.name,
"brand": comp_food.brand,
"unit_type": comp_food.unit_type,
"calories_per_unit": comp_food.calories_per_unit,
"serving_size_g": comp_food.serving_size_g,
"serving_name": comp_food.serving_name,
"is_meal": comp_food.is_meal,
"deleted_at": comp_food.deleted_at,
},
))
meal_data["components"] = comps
meal_data["computed_nutrition_per_meal"] = nutrition.entry_nutrition(meal, 1.0)
return MealRead.model_validate(meal_data)
# ── Meal resolution (for foods.py get_food) ──────────────────────────────────
def load_meal_with_components(db: Session, meal_id: int) -> Food | None:
"""Eager-load a meal food with its components and their foods.
Returns None if the food doesn't exist or is not a meal.
"""
result = db.scalar(
select(Food)
.where(Food.id == meal_id, Food.is_meal.is_(True))
.options(
selectinload(Food.components).joinedload(MealComponent.food)
)
)
return result
# ── Cycle detection ──────────────────────────────────────────────────────────
class MealCycleError(Exception):
"""Raised when a component update would create a cycle in the meal graph."""
def __init__(self, meal_id: int, food_id: int):
super().__init__(
f"Adding food {food_id} as a component of meal {meal_id} "
f"would create a cycle"
)
self.meal_id = meal_id
self.food_id = food_id
def _check_cycle(db: Session, meal_id: int, proposed_food_ids: list[int]) -> None:
"""Raise MealCycleError if adding proposed_food_ids as components of
meal_id would create a cycle (directly or transitively).
Direct self-reference (food_id == meal_id) is also considered a cycle.
"""
# Build adjacency from all existing meal_components
stmt = select(MealComponent.meal_id, MealComponent.food_id)
rows = db.execute(stmt).all()
adj: dict[int, set[int]] = {}
for row in rows:
adj.setdefault(row.meal_id, set()).add(row.food_id)
# Add proposed edges
for fid in proposed_food_ids:
adj.setdefault(meal_id, set()).add(fid)
# DFS from each proposed food_id; if meal_id is reachable, it's a cycle
for fid in proposed_food_ids:
if _dfs_reachable(adj, fid, meal_id, set()):
raise MealCycleError(meal_id, fid)
def _dfs_reachable(
adj: dict[int, set[int]], current: int, target: int, visited: set[int],
) -> bool:
"""Return True if target is reachable from current in the adjacency graph."""
if current == target:
return True
if current in visited:
return False
visited.add(current)
for neighbor in adj.get(current, set()):
if _dfs_reachable(adj, neighbor, target, visited):
return True
return False
# ── create_meal_from_log (§4.3) ──────────────────────────────────────────────
def create_meal_from_log(
db: Session, lookup_date: date, entry_ids: list[int], name: str,
) -> tuple[FoodRead, LogEntryRead]:
"""Create a meal from selected daily_log entries. Runs as one transaction.
Steps:
1. Load the source entries; validate they all exist and belong to ``date``.
2. Create a new Food (is_meal=True, source="meal").
3. Create MealComponent rows (quantity = original entry.quantity).
4. Delete the source entries.
5. Insert ONE replacement entry with quantity=1.0.
6. Commit once; roll back entirely on any failure.
Returns (FoodRead of the new meal, LogEntryRead of the replacement entry).
"""
# 1. Load & validate source entries
source_entries: list[DailyLogEntry] = []
for eid in entry_ids:
entry = db.get(DailyLogEntry, eid)
if entry is None:
raise EntryNotFoundError(eid)
if entry.date != lookup_date:
raise EntryDateMismatchError(eid, entry.date, lookup_date)
source_entries.append(entry)
if not source_entries:
raise ValueError("No valid source entries provided")
# 2. Create the meal food
now = _now()
meal = Food(
name=name,
brand=None,
barcode=None,
source="meal",
is_meal=True,
unit_type="weight", # placeholder — meals use quantity as scaling factor
calories_per_unit=None,
protein_per_unit=None,
carbs_per_unit=None,
fat_per_unit=None,
fiber_per_unit=None,
saturated_fat_per_unit=None,
sugars_per_unit=None,
sodium_per_unit=None,
serving_size_g=None,
serving_name=None,
off_data=None,
deleted_at=None,
created_at=now,
updated_at=now,
)
db.add(meal)
db.flush() # get meal.id
# 3. Create MealComponent rows
for entry in source_entries:
mc = MealComponent(
meal_id=meal.id,
food_id=entry.food_id,
quantity=entry.quantity,
)
db.add(mc)
# 4. Delete source entries
max_sort = 0
meal_slots: set[str | None] = set()
for entry in source_entries:
if entry.sort_order > max_sort:
max_sort = entry.sort_order
meal_slots.add(entry.meal_slot)
db.delete(entry)
# 5. Insert replacement entry
# meal_slot: use the first source entry's slot if all agree, else None
replacement_slot = source_entries[0].meal_slot if len(meal_slots) == 1 else None
replacement = DailyLogEntry(
date=lookup_date,
food_id=meal.id,
quantity=1.0,
meal_slot=replacement_slot,
sort_order=max_sort,
created_at=now,
)
db.add(replacement)
db.flush() # get replacement.id
# Commit
db.commit()
# Re-query for response with eager-loaded relationships
meal_loaded = load_meal_with_components(db, meal.id)
entry_loaded = _load_entry_with_food_components(db, replacement.id)
meal_read = _build_meal_read(meal_loaded)
entry_read = _build_log_entry_read(entry_loaded)
return meal_read, entry_read
# ── unpack_meal (§4.4) ──────────────────────────────────────────────────────
def unpack_meal(
db: Session, meal_id: int, lookup_date: date, entry_id: int | None = None,
) -> list[LogEntryRead]:
"""Replace a logged meal entry with its component entries.
Runs as one transaction:
1. Find the target log entry (by entry_id or auto-detect).
2. Flatten the meal recursively to LEAF foods.
3. Insert new entries for each leaf.
4. Delete the original meal entry.
5. Commit once; roll back entirely on any failure.
If ``entry_id`` is given, it must point at a log entry for ``meal_id``
on ``lookup_date``. If ``entry_id`` is None, the entry is auto-detected:
exactly one matching entry must exist, or an error is raised.
"""
# 1. Find the target log entry
if entry_id is not None:
target = db.scalar(
select(DailyLogEntry)
.where(
DailyLogEntry.id == entry_id,
DailyLogEntry.food_id == meal_id,
DailyLogEntry.date == lookup_date,
)
.options(
joinedload(DailyLogEntry.food)
.selectinload(Food.components)
.joinedload(MealComponent.food)
)
)
if target is None:
raise EntryNotFoundError(entry_id)
else:
candidates = db.scalars(
select(DailyLogEntry)
.where(
DailyLogEntry.food_id == meal_id,
DailyLogEntry.date == lookup_date,
)
.options(
joinedload(DailyLogEntry.food)
.selectinload(Food.components)
.joinedload(MealComponent.food)
)
).unique().all()
if len(candidates) == 0:
raise ValueError(
f"No log entry found for meal {meal_id} on {lookup_date}"
)
if len(candidates) > 1:
raise AmbiguousMealEntryError(meal_id, lookup_date, len(candidates))
target = candidates[0]
meal_food = target.food
if not meal_food.is_meal:
raise ValueError(f"Food {meal_id} is not a meal")
# 2. Flatten recursively to leaf foods
leaf_entries = _flatten_meal(meal_food, target.quantity)
# 3. Insert new entries for each leaf
sort_base = target.sort_order
new_entries: list[DailyLogEntry] = []
for i, (leaf_food_id, eff_qty) in enumerate(leaf_entries):
entry = DailyLogEntry(
date=lookup_date,
food_id=leaf_food_id,
quantity=eff_qty,
meal_slot=target.meal_slot,
sort_order=sort_base + i,
created_at=_now(),
)
db.add(entry)
new_entries.append(entry)
# 4. Delete the original meal entry
db.delete(target)
# Commit
db.commit()
# Re-query for responses with eager-loaded food
result: list[LogEntryRead] = []
for entry in new_entries:
loaded = _load_entry_with_food(db, entry.id)
result.append(_build_log_entry_read(loaded))
return result
def _flatten_meal(
food: Food, scaling: float, visited: set[int] | None = None,
) -> list[tuple[int, float]]:
"""Recursively flatten a meal to its leaf foods.
Returns a list of (food_id, effective_quantity) for each leaf food.
Nested meals are expanded; their scaling factors are multiplied down.
``visited`` guards against cycles (safety net; cycles should be prevented
by the write path). A cycled branch returns an empty list.
"""
if visited is None:
visited = set()
if food.id in visited:
return []
visited.add(food.id)
if not food.is_meal:
return [(food.id, scaling)]
leaves: list[tuple[int, float]] = []
for component in food.components:
# component.quantity is the amount in ONE full meal; multiply by
# the scaling factor for this log entry.
leaves.extend(
_flatten_meal(component.food, component.quantity * scaling, visited)
)
return leaves
# ── update_meal_components (§3.2 PUT, cycle-checked) ─────────────────────────
def update_meal_components(
db: Session, meal_id: int, components: list[dict],
) -> MealRead:
"""Replace the component list for a meal food.
``components`` is a list of dicts with ``food_id`` and ``quantity`` keys.
Validates that each food_id exists, quantity > 0, and the new component
list doesn't create a cycle (direct or transitive self-reference).
Runs in one transaction: deletes existing components, inserts new ones,
commits. Returns the updated MealRead.
"""
# Validate meal exists and is a meal
meal = db.get(Food, meal_id)
if meal is None or not meal.is_meal:
raise MealNotFoundError(meal_id)
# Collect proposed food_ids
proposed_ids = [c["food_id"] for c in components]
# Validate each food_id exists
for fid in proposed_ids:
if db.get(Food, fid) is None:
raise ComponentFoodNotFoundError(fid)
# Cycle check: would adding these components create a cycle?
_check_cycle(db, meal_id, proposed_ids)
# Replace: delete existing, insert new
db.execute(
MealComponent.__table__.delete().where(MealComponent.meal_id == meal_id)
)
for c in components:
mc = MealComponent(
meal_id=meal_id,
food_id=c["food_id"],
quantity=c["quantity"],
)
db.add(mc)
db.commit()
# Re-query with components loaded
meal = load_meal_with_components(db, meal_id)
return _build_meal_read(meal)
# ── Errors ───────────────────────────────────────────────────────────────────
class EntryNotFoundError(Exception):
def __init__(self, entry_id: int):
super().__init__(f"Log entry {entry_id} not found")
self.entry_id = entry_id
class EntryDateMismatchError(Exception):
def __init__(self, entry_id: int, entry_date: date, expected_date: date):
super().__init__(
f"Log entry {entry_id} has date {entry_date}, "
f"not the expected {expected_date}"
)
self.entry_id = entry_id
self.entry_date = entry_date
self.expected_date = expected_date
class AmbiguousMealEntryError(Exception):
def __init__(self, meal_id: int, lookup_date: date, count: int):
super().__init__(
f"Multiple ({count}) log entries for meal {meal_id} on {lookup_date}. "
f"Specify entry_id to disambiguate."
)
self.meal_id = meal_id
self.lookup_date = lookup_date
self.count = count
class MealNotFoundError(Exception):
def __init__(self, meal_id: int):
super().__init__(f"Meal {meal_id} not found or is not a meal")
self.meal_id = meal_id
class ComponentFoodNotFoundError(Exception):
def __init__(self, food_id: int):
super().__init__(f"Component food {food_id} not found")
self.food_id = food_id
backend/services/nutrition.py
+67 -19
View File
@@ -5,6 +5,10 @@ Routers never compute nutrition; the frontend never re-derives it.
- unit_type "weight": quantity is grams; nutrition = (quantity / 100) × per_unit
- unit_type "count": quantity is item count; nutrition = quantity × per_unit
- meal: quantity is a scaling factor (1.0 = one full meal)
Meal nutrition is derived by recursive component summation (§2.2), with cycle
detection. A cycle returns zeros for that branch (safety net — writes prevent
cycles, but reads must never infinite-loop).
"""
from models import Food
@@ -42,31 +46,75 @@ def scale_to_quantity(per_unit: float | None, quantity: float, unit_type: str) -
raise ValueError(f"unknown unit_type: {unit_type!r}")
def entry_calories(food: Food, quantity: float) -> float:
"""Calories for a single (non-meal) food at a logged quantity.
def entry_calories(food: Food, quantity: float, visited: set[int] | None = None) -> float:
"""Calories for a food at a logged quantity.
TODO: handle is_meal foods by summing scaled component nutrition
(recursively, with cycle detection — spec §2.2).
For regular foods: scales per_unit by quantity using unit_type.
For meals: recursively sums scaled component nutrition. Cycle detection
prevents infinite loops — a cycled branch returns 0 (safety net; writes
should prevent cycles from being created).
"""
return scale_to_quantity(food.calories_per_unit, quantity, food.unit_type)
if visited is None:
visited = set()
if food.id in visited:
return 0.0
if not food.is_meal:
return scale_to_quantity(food.calories_per_unit, quantity, food.unit_type)
# Meal: recurse into components
visited.add(food.id)
total = 0.0
for component in food.components:
# component.quantity is the amount in ONE full meal; multiply by the
# entry's scaling factor to get the effective quantity for this log entry.
total += entry_calories(component.food, component.quantity * quantity, visited)
return total
def entry_nutrition(food: Food, quantity: float) -> dict[str, float]:
def entry_nutrition(
food: Food, quantity: float, visited: set[int] | None = None,
) -> dict[str, float]:
"""Return all nutrition fields scaled to a logged quantity.
Each field is resolved via scale_to_quantity using the food's unit_type.
NULL per-unit values contribute 0.0, not an error.
For regular foods: each field is resolved via scale_to_quantity using
the food's unit_type. NULL per-unit values contribute 0.0.
Meal foods (is_meal=True) have null per_unit fields per the CHECK
constraint, so they naturally contribute 0 for all fields.
TODO: TICKET-007 — real meal nutrition will sum scaled component
foods recursively. Until then, meal entries contribute 0.
For meals: recursively sums scaled component nutrition (§2.2). Cycle
detection prevents infinite loops — a cycled branch returns zeros for
all fields (safety net; writes should prevent cycles from being created).
The ``visited`` set tracks food IDs on the current recursion path.
Callers should NOT pre-populate it — it defaults to an empty set and
is only used internally for recursion.
"""
return {
field: scale_to_quantity(
getattr(food, _FIELD_TO_PER_UNIT_COL[field], None),
quantity,
food.unit_type,
if visited is None:
visited = set()
if food.id in visited:
# Cycle detected — safety net; return zeros for this branch.
return {field: 0.0 for field in NUTRITION_FIELDS}
if not food.is_meal:
return {
field: scale_to_quantity(
getattr(food, _FIELD_TO_PER_UNIT_COL[field], None),
quantity,
food.unit_type,
)
for field in NUTRITION_FIELDS
}
# Meal: recurse into components, multiplying the scaling factor down.
visited.add(food.id)
totals = {field: 0.0 for field in NUTRITION_FIELDS}
for component in food.components:
# component.quantity is the amount in ONE full meal; multiply by the
# entry's scaling factor to get the effective quantity for this log entry.
component_nut = entry_nutrition(
component.food, component.quantity * quantity, visited,
)
for field in NUTRITION_FIELDS
}
for field in NUTRITION_FIELDS:
totals[field] += component_nut[field]
return totals