Craig
552 lines
18 KiB
Python
552 lines
18 KiB
Python
"""Meal business logic: composition, recursion, cycle detection (spec §2.2).
|
|
|
|
Multi-write operations here own their transactions (spec §8.1 rule 6):
|
|
create_meal_from_log, unpack_meal, and update_meal_components each commit once
|
|
at the end or roll back entirely — never partial writes.
|
|
|
|
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
|