"""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