feat(ingest): persist every source through the repository layer (#36)

.gitignore

@@ -50,3 +50,6 @@ backend/data/
 
 # Personal developer journal (private, not for team repo)
 .journal-entries/
+
+# requests-cache store written by the Open-Meteo client
+.cache.sqlite

README.md

@@ -83,6 +83,7 @@ foresight/
 │   │   ├── services/            # Business logic
 │   │   │   ├── ml/              # ML models (regression, bayesian, ...)
 │   │   │   └── parsers/         # Client data ingestion
+│   │   ├── ingest/              # External-source adapters + CLIs (see docs/)
 │   │   ├── repository/          # Data-access layer
 │   │   └── database/            # SQLAlchemy + Alembic
 │   │       ├── base.py          # Declarative base

agents/repository.md

@@ -14,13 +14,16 @@ returns models/data back up. No business rules, no HTTP.
 
 ## Patterns & conventions
 - Functions are `async` and receive an `AsyncSession` (don't open their own).
+- **Never commit.** The caller owns the transaction: `get_session()` commits
+  once per request; CLIs and jobs commit once at the end. Use `flush()` when you
+  need generated ids.
 - Encapsulate all query/CRUD logic here so services stay persistence-agnostic.
 - Return ORM models or simple data; let callers map to schemas.
 
 ## Rules / invariants
 - Only this layer imports `session` / ORM query APIs.
 - No business logic and no HTTP concerns.
-- Don't manage transactions ad hoc — use the session provided via `get_session()`.
+- No `commit()` or `rollback()` here — that's the caller's job.
 
 ## Reference example
 Copy the pattern in `backend/foresight/repository/example_widget.py` (the

agents/routes.md

@@ -16,6 +16,9 @@ and no database access here.
 ## Patterns & conventions
 - One `APIRouter` per module, with `tags=[...]`; include it in `main.py`.
 - Handlers are `async def`. Type request bodies and responses with schemas.
+- Inject the session as `Depends(get_session, scope="function")` (see
+  `SessionDep` in the widget example). It commits once per request after the
+  handler returns; handlers and services never commit.
 - Call into services; translate service results into responses. Keep handlers thin.
 
 ## Rules / invariants

agents/services.md

@@ -11,6 +11,9 @@ domain rules live — they're called by routes and call the repository.
 - `backend/foresight/services/` — business logic modules.
 - `backend/foresight/services/ml/` — modeling / recommendation logic.
 - `backend/foresight/services/parsers/` — ingesting external + hotel data.
+- `backend/foresight/ingest/` — external-source adapters (events, weather,
+  flights). Part of the services layer: they fetch and normalize, then write
+  through `repository/`.
 
 ## Patterns & conventions
 - Pure-ish domain functions/classes; take plain inputs or schemas, return domain
@@ -22,6 +25,8 @@ domain rules live — they're called by routes and call the repository.
 - No FastAPI/HTTP concerns (no `Request`, no status codes, no routers).
 - No direct ORM/session use — all persistence goes through `repository/`.
 - External API access and heavy logic belong here, not in routes.
+- Never open a session or commit. CLIs live in `backend/scripts/`; they open the
+  session, call the service, and commit once.
 
 ## Reference example
 Copy the pattern in `backend/foresight/services/example_widget.py`.

backend/foresight/database/migrations/versions/b9719be9df40_external_signal_upsert_keys.py

@@ -0,0 +1,57 @@
+"""external signal upsert keys
+
+Gives weather, flight_arrivals and competitor_rates the natural keys their
+ingestion needs: without a conflict target every re-run appends duplicate
+rows. Source stays in each key, so two providers describing the same day are
+distinct rows rather than silently overwriting one another, and a provider
+re-fetching a day updates its own row in place.
+
+Revision ID: b9719be9df40
+Revises: a1b2c3d4e5f6
+Create Date: 2026-10-12 02:26:03.496929
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'b9719be9df40'
+down_revision: Union[str, None] = 'a1b2c3d4e5f6'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_unique_constraint(
+        'uq_weather_city_date_kind_source',
+        'weather',
+        ['city', 'weather_date', 'is_forecast', 'source'],
+    )
+    op.create_unique_constraint(
+        'uq_flight_arrivals_airport_date_source',
+        'flight_arrivals',
+        ['airport', 'arrival_date', 'source'],
+    )
+    op.create_unique_constraint(
+        'uq_competitor_rates_hotel_date_source',
+        'competitor_rates',
+        ['hotel_name', 'city', 'stay_date', 'source'],
+    )
+    # The unique constraint above leads on hotel_name, so it cannot serve the
+    # market-wide read: every competitor's rate across a range of nights.
+    op.create_index(
+        'ix_competitor_rates_city_stay', 'competitor_rates', ['city', 'stay_date']
+    )
+
+
+def downgrade() -> None:
+    op.drop_index('ix_competitor_rates_city_stay', table_name='competitor_rates')
+    op.drop_constraint(
+        'uq_competitor_rates_hotel_date_source', 'competitor_rates', type_='unique'
+    )
+    op.drop_constraint(
+        'uq_flight_arrivals_airport_date_source', 'flight_arrivals', type_='unique'
+    )
+    op.drop_constraint('uq_weather_city_date_kind_source', 'weather', type_='unique')

backend/foresight/database/migrations/versions/c97a8a991e88_hotel_upsert_keys.py

@@ -0,0 +1,46 @@
+"""hotel upsert keys
+
+Gives hotels, room_types and daily_performance the named conflict targets
+repository/hotel.py already upserts against; without them every one of its
+three functions raises. daily_performance only needs the rename, since its
+unique constraint was created unnamed and Postgres generated the name.
+
+Revision ID: c97a8a991e88
+Revises: b9719be9df40
+Create Date: 2026-10-12 21:42:11.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'c97a8a991e88'
+down_revision: Union[str, None] = 'b9719be9df40'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+_GENERATED_PERF_KEY = 'daily_performance_room_type_id_stay_date_key'
+
+
+def upgrade() -> None:
+    op.create_unique_constraint('uq_hotels_name_city', 'hotels', ['name', 'city'])
+    op.create_unique_constraint(
+        'uq_room_types_hotel_name', 'room_types', ['hotel_id', 'name']
+    )
+    # A rename, not a drop and recreate, so the table is never without its
+    # unique and the existing index is reused.
+    op.execute(
+        f'ALTER TABLE daily_performance RENAME CONSTRAINT '
+        f'{_GENERATED_PERF_KEY} TO uq_daily_performance_room_type_stay'
+    )
+
+
+def downgrade() -> None:
+    op.execute(
+        f'ALTER TABLE daily_performance RENAME CONSTRAINT '
+        f'uq_daily_performance_room_type_stay TO {_GENERATED_PERF_KEY}'
+    )
+    op.drop_constraint('uq_room_types_hotel_name', 'room_types', type_='unique')
+    op.drop_constraint('uq_hotels_name_city', 'hotels', type_='unique')

backend/foresight/database/models/external.py

@@ -52,6 +52,18 @@ class Weather(Base):
         DateTime(timezone=True), server_default=func.now()
     )
 
+    # A forecast and the recorded actual for the same day are different rows, as
+    # are two sources' readings; re-fetching either one updates in place.
+    __table_args__ = (
+        UniqueConstraint(
+            "city",
+            "weather_date",
+            "is_forecast",
+            "source",
+            name="uq_weather_city_date_kind_source",
+        ),
+    )
+
 
 class Event(Base):
     """One real-world event occurrence, merged across sources.
@@ -176,6 +188,19 @@ class CompetitorRate(Base):
         DateTime(timezone=True), server_default=func.now()
     )
 
+    __table_args__ = (
+        UniqueConstraint(
+            "hotel_name",
+            "city",
+            "stay_date",
+            "source",
+            name="uq_competitor_rates_hotel_date_source",
+        ),
+        # The unique constraint leads on hotel_name, so it can't serve the
+        # market-wide read: every competitor's rate for a range of nights.
+        Index("ix_competitor_rates_city_stay", "city", "stay_date"),
+    )
+
 
 class FlightArrival(Base):
     """How many people are flying in on a given day."""
@@ -193,3 +218,14 @@ class FlightArrival(Base):
     captured_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now()
     )
+
+    # CSO, OpenSky and AeroDataBox measure the same day at different grains, so
+    # each keeps its own row and the read layer picks which one wins.
+    __table_args__ = (
+        UniqueConstraint(
+            "airport",
+            "arrival_date",
+            "source",
+            name="uq_flight_arrivals_airport_date_source",
+        ),
+    )

backend/foresight/database/models/hotel.py

@@ -23,6 +23,8 @@ class Hotel(Base):
 
     room_types: Mapped[list["RoomType"]] = relationship(back_populates="hotel")
 
+    __table_args__ = (UniqueConstraint("name", "city", name="uq_hotels_name_city"),)
+
 
 class RoomType(Base):
     """A category of room, e.g. "Deluxe King". Price is set per room type."""
@@ -38,6 +40,10 @@ class RoomType(Base):
     hotel: Mapped["Hotel"] = relationship(back_populates="room_types")
     nights: Mapped[list["DailyPerformance"]] = relationship(back_populates="room_type")
 
+    __table_args__ = (
+        UniqueConstraint("hotel_id", "name", name="uq_room_types_hotel_name"),
+    )
+
 
 class DailyPerformance(Base):
     """One room type, one night: what we charged and how it sold."""
@@ -56,4 +62,8 @@ class DailyPerformance(Base):
     room_type: Mapped["RoomType"] = relationship(back_populates="nights")
 
     # One row per room type per night.
-    __table_args__ = (UniqueConstraint("room_type_id", "stay_date"),)
+    __table_args__ = (
+        UniqueConstraint(
+            "room_type_id", "stay_date", name="uq_daily_performance_room_type_stay"
+        ),
+    )

backend/foresight/database/session.py

@@ -17,5 +17,16 @@
 
 
 async def get_session() -> AsyncGenerator[AsyncSession, None]:
+    """One transaction per request: commits on success, rolls back on error.
+
+    Repositories never commit. Depend on this with `scope="function"` so the
+    commit runs before the response is sent and a failed commit returns a 500.
+    """
     async with async_session_factory() as session:
-        yield session
+        try:
+            yield session
+        except Exception:
+            await session.rollback()
+            raise
+        else:
+            await session.commit()

backend/foresight/ingest/external.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import argparse
+import asyncio
+import sys
+from datetime import date
+from typing import TYPE_CHECKING
+
+from foresight.database.session import async_session_factory, engine
+from foresight.ingest.flights import daily_arrivals
+from foresight.ingest.weather import fetch_actuals, fetch_forecast
+from foresight.repository.flights import upsert_flight_arrivals
+from foresight.repository.weather import upsert_weather
+from foresight.services.flight_data_service import fetch_aerodatabox, fetch_opensky
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    from foresight.repository import Outcome
+
+# shortcut: every flight provider is hardcoded to Dublin, so there is nothing
+# for an --airport flag to vary yet; add one when a second airport appears.
+AIRPORT = "DUB"
+_FLIGHT_PROVIDERS = {"opensky": fetch_opensky, "aerodatabox": fetch_aerodatabox}
+
+
+def _parse_date(value: str) -> date:
+    try:
+        return date.fromisoformat(value)
+    except ValueError:
+        raise argparse.ArgumentTypeError(
+            f"expected a date like 2026-10-05, got {value!r}"
+        ) from None
+
+
+async def _weather(
+    session: AsyncSession, args: argparse.Namespace
+) -> dict[Outcome, int]:
+    if args.start is None:
+        rows = fetch_forecast(args.city, args.lat, args.lon, "auto")
+    else:
+        # The archive refuses a future end date outright instead of returning
+        # empty days, so a range running past today has to be clamped here.
+        rows = fetch_actuals(
+            args.city,
+            args.lat,
+            args.lon,
+            "auto",
+            start=args.start,
+            end=min(args.end, date.today()),
+        )
+    return await upsert_weather(session, rows)
+
+
+async def _flights(
+    session: AsyncSession, args: argparse.Namespace
+) -> dict[Outcome, int]:
+    frame = _FLIGHT_PROVIDERS[args.source](args.start, args.end)
+    rows = daily_arrivals(frame, airport=AIRPORT, source=args.source)
+    return await upsert_flight_arrivals(session, rows)
+
+
+async def run(args: argparse.Namespace) -> dict[Outcome, int]:
+    handler = _weather if args.source == "weather" else _flights
+    async with async_session_factory() as session:
+        counts = await handler(session, args)
+        await session.commit()
+    return counts
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="python -m foresight.ingest.external",
+        description="Fetch one external source and upsert it into its table.",
+    )
+    parser.add_argument(
+        "--source", choices=["weather", *_FLIGHT_PROVIDERS], required=True
+    )
+    parser.add_argument("--start", type=_parse_date, help="first day, inclusive")
+    parser.add_argument("--end", type=_parse_date, help="last day, inclusive")
+    parser.add_argument("--city", help="weather only: the name stored on each row")
+    parser.add_argument("--lat", type=float, help="weather only")
+    parser.add_argument("--lon", type=float, help="weather only")
+    return parser
+
+
+async def _main(args: argparse.Namespace) -> dict[Outcome, int]:
+    try:
+        return await run(args)
+    finally:
+        await engine.dispose()
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = _parser()
+    args = parser.parse_args(argv)
+    if args.source == "weather":
+        if None in (args.city, args.lat, args.lon):
+            parser.error("--source weather needs --city, --lat and --lon")
+        if (args.start is None) != (args.end is None):
+            parser.error("give both --start and --end, or neither for a forecast")
+    elif None in (args.start, args.end):
+        parser.error(f"--source {args.source} needs --start and --end")
+    if None not in (args.start, args.end) and args.start > args.end:
+        parser.error("--start is after --end")
+
+    engine.echo = False
+    counts = asyncio.run(_main(args))
+    total = sum(counts.values())
+    print(
+        f"{args.source}: {total} rows "
+        f"({', '.join(f'{n} {outcome.value}' for outcome, n in counts.items())})"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())