"""Targets service layer — business logic + DB access (spec §8.1 rules 2–3, 6). Auto-close semantics: - When a new target is created, the previous active target's end_date is set to the new target's start_date (same day). - Historical lookup uses an EXCLUSIVE end_date: a target covers dates where start_date <= date AND (end_date IS NULL OR end_date > date). This forms half-open intervals [start_date, end_date), so there is never overlap or ambiguity about which target applies on a boundary date. ORM objects never leave this module; functions return Pydantic schemas. """ from datetime import date from sqlalchemy import select from sqlalchemy.orm import Session from models import Target from schemas import TargetCreate, TargetRead, TargetUpdate # ── CRUD ───────────────────────────────────────────────────────────────────── def list_targets(db: Session) -> list[TargetRead]: """List all targets ordered by start_date.""" targets = db.scalars(select(Target).order_by(Target.start_date)).all() return [TargetRead.model_validate(t) for t in targets] def get_current_target(db: Session) -> TargetRead | None: """Return the active target (end_date IS NULL), or None.""" target = db.scalar(select(Target).where(Target.end_date.is_(None))) if target is None: return None return TargetRead.model_validate(target) def create_target(db: Session, data: TargetCreate) -> TargetRead: """Create a new target. Auto-closes the previous active target by setting its end_date to the new target's start_date. Runs in a single transaction so there is never a window with zero or two active targets (§2.4, §8.1 rule 6). Raises ValueError if the auto-close would cause the previous target's end_date to be <= its start_date (e.g. backdating a target before the current one started). """ new_start = data.start_date # Find the currently active target and auto-close it current_active = db.scalar( select(Target).where(Target.end_date.is_(None)) ) if current_active is not None: if new_start <= current_active.start_date: raise ValueError( f"New target start_date ({new_start}) is not after the current " f"active target's start_date ({current_active.start_date}). " f"Cannot auto-close without creating an invalid date range." ) current_active.end_date = new_start target = Target( start_date=data.start_date, end_date=None, calories=data.calories, protein_g=data.protein_g, carbs_g=data.carbs_g, fat_g=data.fat_g, ) db.add(target) db.commit() db.refresh(target) return TargetRead.model_validate(target) def update_target(db: Session, target_id: int, data: TargetUpdate) -> TargetRead | None: """Update a target's values and/or date range. The single-active-target invariant is preserved: setting end_date to null when another active target already exists is rejected. Returns None if the target is not found. Raises ValueError for date-range validation failures. Raises ActiveTargetConflictError if the update would create two active targets. """ target = db.get(Target, target_id) if target is None: return None update_data = data.model_dump(exclude_unset=True) # Validate the resulting date range if either date field is being changed new_start = update_data.get("start_date", target.start_date) new_end = update_data.get("end_date", target.end_date) # None means "not being updated" # If end_date key is present in update_data, use that value (could be None explicitly) if "end_date" in update_data: new_end = update_data["end_date"] else: new_end = target.end_date # Validate end_date > start_date when both are set if new_end is not None and new_end <= new_start: raise ValueError("end_date must be after start_date") # Check single-active-target invariant: if setting end_date to NULL, # there must not already be another active target if "end_date" in update_data and update_data["end_date"] is None: if target.end_date is not None: # target is currently NOT active other_active = db.scalar( select(Target).where( Target.end_date.is_(None), Target.id != target_id, ) ) if other_active is not None: raise ActiveTargetConflictError(other_active.id) # If changing start_date on the active target, ensure it doesn't break # the invariant with respect to the previously-closed target. # (The previously-closed target's end_date references the OLD start_date; # changing start_date on the active one doesn't retroactively fix that. # This is fine — historical targets may have end_date values that don't # align perfectly after edits. The half-open interval lookup still works.) for field, value in update_data.items(): setattr(target, field, value) db.commit() db.refresh(target) return TargetRead.model_validate(target) # ── Historical lookup ──────────────────────────────────────────────────────── def get_target_for_date(db: Session, lookup_date: date) -> TargetRead | None: """Return the target whose date range contains the given date. Uses half-open intervals [start_date, end_date): a target covers dates where start_date <= date AND (end_date IS NULL OR end_date > date). Returns None if no target covers the given date. Exposed for TICKET-004 (day summary endpoint). """ target = db.scalar( select(Target) .where( Target.start_date <= lookup_date, (Target.end_date.is_(None)) | (Target.end_date > lookup_date), ) .order_by(Target.start_date.desc()) .limit(1) ) if target is None: return None return TargetRead.model_validate(target) # ── Errors ─────────────────────────────────────────────────────────────────── class ActiveTargetConflictError(Exception): """Raised when an operation would create two active targets.""" def __init__(self, existing_active_id: int): super().__init__( f"There is already an active target (id={existing_active_id}). " f"Close it first before activating another." ) self.existing_active_id = existing_active_id