@@ -0,0 +1,46 @@
	1	+name: Eurostat benchmark (monthly)
	2	+
	3	+# Official monthly passengers per EU airport, published ~2 months behind. One
	4	+# request per month of the window; the parquet is committed so every lake run
	5	+# and dbt build sees it without another download.
	6	+on:
	7	+ schedule:
	8	+ - cron: "0 4 1 * *"
	9	+ workflow_dispatch:
	10	+
	11	+permissions:
	12	+ contents: write
	13	+
	14	+concurrency:
	15	+ group: eurostat
	16	+ cancel-in-progress: true
	17	+
	18	+jobs:
	19	+ fetch:
	20	+ name: Fetch Eurostat airport traffic
	21	+ runs-on: ubuntu-latest
	22	+ timeout-minutes: 15
	23	+ steps:
	24	+ - uses: actions/checkout@v4
	25	+
	26	+ - name: Install uv
	27	+ uses: astral-sh/setup-uv@v5
	28	+
	29	+ - name: Set up Python
	30	+ uses: actions/setup-python@v5
	31	+ with:
	32	+ python-version: "3.13"
	33	+
	34	+ - name: Install dependencies
	35	+ run: uv sync --extra dev
	36	+
	37	+ - name: Fetch
	38	+ run: uv run python -m scripts.fetch_eurostat
	39	+
	40	+ - name: Commit the benchmark file
	41	+ run: \|
	42	+ git config user.name "github-actions[bot]"
	43	+ git config user.email "github-actions[bot]@users.noreply.github.com"
	44	+ git add services/data/eurostat_airport_traffic.parquet
	45	+ git diff --cached --quiet \|\| git commit -m "chore: monthly Eurostat airport traffic"
	46	+ git push
0	47

▼dbt/models/eurostat_sources.yml

+9-0

@@ -0,0 +1,9 @@
	1	+version: 2
	2	+
	3	+sources:
	4	+ - name: eurostat
	5	+ description: "Official Eurostat statistics fetched monthly by GitHub Actions"
	6	+ schema: main
	7	+ tables:
	8	+ - name: eurostat_airport_traffic
	9	+ description: "Monthly passengers per EU airport (avia_paoa, ~2 month lag)"
0	10

▼dbt/models/marts/airport_official_traffic.sql

+43-0

@@ -0,0 +1,43 @@
	1	+{{ config(materialized='view', tags=['marts']) }}
	2	+
	3	+with official as (
	4	+ select * from {{ ref('stg_eurostat_airport_traffic') }}
	5	+),
	6	+
	7	+airports as (
	8	+ select airport_icao, name as airport_name, iso_country as country from {{ ref('stg_dim_airport') }}
	9	+),
	10	+
	11	+latest as (
	12	+ select max(period_start) as period_start from official
	13	+),
	14	+
	15	+metrics as (
	16	+ select
	17	+ o.airport_icao,
	18	+ a.airport_name,
	19	+ a.country,
	20	+ max(o.period_start) as latest_month,
	21	+ sum(case when o.period_start >= l.period_start - interval '12 months' then o.passengers end)
	22	+ as passengers_12m,
	23	+ sum(case when o.period_start >= date_trunc('year', l.period_start) then o.passengers end)
	24	+ as passengers_ytd,
	25	+ count(distinct o.period) as months_reported
	26	+ from official o
	27	+ cross join latest l
	28	+ left join airports a on o.airport_icao = a.airport_icao
	29	+ group by 1, 2, 3
	30	+)
	31	+
	32	+select
	33	+ airport_icao,
	34	+ airport_name,
	35	+ country,
	36	+ latest_month,
	37	+ passengers_12m,
	38	+ passengers_ytd,
	39	+ months_reported,
	40	+ row_number() over (order by passengers_12m desc nulls last) as official_rank
	41	+from metrics
	42	+where passengers_12m is not null
	43	+order by official_rank
0	44

▼dbt/models/staging/schema.yml

+14-0

@@ -160,3 +160,17 @@ models:
160	160	values: ['runway_closure', 'taxiway_closure', 'navaid', 'obstacle',
161	161	'airport_closure', 'fire', 'bird_activity', 'weather',
162	162	'airspace', 'other']
	163	+
	164	+ - name: stg_eurostat_airport_traffic
	165	+ description: "Eurostat monthly passengers per airport (official benchmark)"
	166	+ columns:
	167	+ - name: period_start
	168	+ tests:
	169	+ - not_null
	170	+ - name: airport_icao
	171	+ tests:
	172	+ - not_null
	173	+ - name: passengers
	174	+ tests:
	175	+ - not_null
	176	+
163	177

▼dbt/models/staging/stg_eurostat_airport_traffic.sql

+16-0

@@ -0,0 +1,16 @@
	1	+{{ config(materialized='view', tags=['staging']) }}
	2	+
	3	+with source as (
	4	+ select * from {{ source('eurostat', 'eurostat_airport_traffic') }}
	5	+)
	6	+
	7	+select
	8	+ period,
	9	+ cast(period \|\| '-01' as date) as period_start,
	10	+ airport_icao,
	11	+ cast(passengers as bigint) as passengers,
	12	+ source,
	13	+ cast(fetched_at as timestamp) as fetched_at
	14	+from source
	15	+where period is not null
	16	+ and airport_icao is not null
0	17

▼pipelines/warehouse.py

+27-0

@@ -17,6 +17,7 @@ dbt views under the ``gold_`` prefix for downstream consumers.
17	17	from __future__ import annotations
18	18
19	19	import datetime as _dt
	20	+from pathlib import Path
20	21	from typing import Any
21	22
22	23	import duckdb
@@ -161,6 +162,16 @@ _EMPTY_DIM_ROUTE = {
161	162	LIVE_POSITION_WINDOW_HOURS = 1
162	163	LIVE_POSITION_MAX_ROWS = 5_000
163	164
	165	+# Eurostat official monthly passengers per airport (benchmark layer).
	166	+_EUROSTAT_TRAFFIC_COLUMNS = ("period", "airport_icao", "passengers", "source", "fetched_at")
	167	+_EMPTY_EUROSTAT_TRAFFIC = {
	168	+ "period": "VARCHAR",
	169	+ "airport_icao": "VARCHAR",
	170	+ "passengers": "BIGINT",
	171	+ "source": "VARCHAR",
	172	+ "fetched_at": "VARCHAR",
	173	+}
	174	+
164	175	_FACT_POSITIONS_COLUMNS = (
165	176	"icao24",
166	177	"callsign",
@@ -416,6 +427,21 @@ class WarehouseBuilder:
416	427	self._load_replace("weather", weather)
417	428	logger.info("[warehouse] weather rows=%s", weather.height)
418	429
	430	+ def build_eurostat_traffic(self) -> None:
	431	+ """Official Eurostat monthly passengers per airport (if fetched)."""
	432	+ path = (
	433	+ Path(self.settings.project_root)
	434	+ / "services"
	435	+ / "data"
	436	+ / "eurostat_airport_traffic.parquet"
	437	+ )
	438	+ if not path.exists():
	439	+ self._create_or_replace_empty("eurostat_airport_traffic", _EMPTY_EUROSTAT_TRAFFIC)
	440	+ return
	441	+ frame = _ensure_columns(pl.read_parquet(path), _EUROSTAT_TRAFFIC_COLUMNS)
	442	+ self._load_replace("eurostat_airport_traffic", frame)
	443	+ logger.info("[warehouse] eurostat_airport_traffic rows=%s", frame.height)
	444	+
419	445	def build_fact_positions(self) -> None:
420	446	"""Latest live aircraft position per airframe (with class + CO₂).
421	447
@@ -525,6 +551,7 @@ class WarehouseBuilder:
525	551	self.build_fact_flights()
526	552	self.build_weather()
527	553	self.build_fact_positions()
	554	+ self.build_eurostat_traffic()
528	555	self.build_fact_emissions()
529	556	self.build_fact_notams()
530	557
531	558

▼scripts/fetch_eurostat.py

+139-0

@@ -0,0 +1,139 @@
	1	+"""Fetch Eurostat monthly airport traffic (official benchmark layer).
	2	+
	3	+`avia_paoa` reports passenger numbers for ~870 European airports, monthly,
	4	+~2 months behind. It is a benchmark, not live data: the Gold layer joins it
	5	+to our counted movements so the airport leaderboard can be cross-checked
	6	+against the official figures.
	7	+
	8	+One request covers every month in the dataset, so this runs at most monthly:
	9	+
	10	+ python -m scripts.fetch_eurostat
	11	+"""
	12	+
	13	+from __future__ import annotations
	14	+
	15	+import argparse
	16	+import json
	17	+from datetime import UTC, datetime
	18	+from pathlib import Path
	19	+from typing import Any
	20	+
	21	+import polars as pl
	22	+import requests
	23	+
	24	+from config.logging import logger
	25	+from config.settings import settings
	26	+
	27	+EUROSTAT_URL = "https://ec.europa.eu/eurostat/api/dissemination/statistics/1.0/data/avia_paoa"
	28	+# Monthly boarded passengers, all schedule types and coverage.
	29	+FILTERS = {
	30	+ "format": "JSON",
	31	+ "lang": "EN",
	32	+ "freq": "M",
	33	+ "unit": "PAS",
	34	+ "tra_meas": "PAS_BRD",
	35	+ "schedule": "TOTAL",
	36	+}
	37	+
	38	+
	39	+def _decode(payload: dict[str, Any]) -> list[dict[str, Any]]:
	40	+ """Decode JSON-stat 2.0 into rows filtered to the requested dimensions."""
	41	+ order: list[str] = payload.get("id") or []
	42	+ size: list[int] = payload.get("size") or []
	43	+ if not order or not size:
	44	+ return []
	45	+ dimensions = payload.get("dimension") or {}
	46	+ categories: dict[str, list[str]] = {}
	47	+ for name in order:
	48	+ index = (dimensions.get(name) or {}).get("category", {}).get("index") or {}
	49	+ if isinstance(index, dict):
	50	+ categories[name] = [code for code, _ in sorted(index.items(), key=lambda kv: kv[1])]
	51	+ else:
	52	+ categories[name] = [str(entry) for entry in index]
	53	+
	54	+ airport_at = order.index("rep_airp")
	55	+ time_at = order.index("time")
	56	+ rows: list[dict[str, Any]] = []
	57	+ for flat, value in (payload.get("value") or {}).items():
	58	+ if value is None:
	59	+ continue
	60	+ remaining = int(flat)
	61	+ indices = [0] * len(order)
	62	+ for position in range(len(order) - 1, -1, -1):
	63	+ indices[position] = remaining % size[position]
	64	+ remaining //= size[position]
	65	+ airport = categories["rep_airp"][indices[airport_at]]
	66	+ period = categories["time"][indices[time_at]]
	67	+ # Eurostat airport codes look like "DE_EDDF"; the ICAO part joins dim_airport.
	68	+ icao = airport.split("_")[-1].upper()
	69	+ if len(icao) != 4 or not icao.isalnum():
	70	+ continue
	71	+ try:
	72	+ passengers = int(float(value))
	73	+ except (TypeError, ValueError):
	74	+ continue
	75	+ rows.append({"period": str(period), "airport_icao": icao, "passengers": passengers})
	76	+ return rows
	77	+
	78	+
	79	+def _recent_months(count: int) -> list[str]:
	80	+ """The last ``count`` calendar months as YYYY-MM (Eurostat rejects bulk pulls)."""
	81	+ today = datetime.now(UTC).date()
	82	+ months: list[str] = []
	83	+ year, month = today.year, today.month
	84	+ for _ in range(count):
	85	+ months.append(f"{year:04d}-{month:02d}")
	86	+ month -= 1
	87	+ if month == 0:
	88	+ year, month = year - 1, 12
	89	+ return list(reversed(months))
	90	+
	91	+
	92	+def fetch(months: int = 12) -> list[dict[str, Any]]:
	93	+ rows: list[dict[str, Any]] = []
	94	+ for period in _recent_months(months):
	95	+ response = requests.get(EUROSTAT_URL, params={**FILTERS, "time": period}, timeout=60)
	96	+ response.raise_for_status()
	97	+ rows.extend(_decode(response.json()))
	98	+ return rows
	99	+
	100	+
	101	+def write_parquet(app_settings=None, months: int = 12) -> Path:
	102	+ cfg = app_settings or settings
	103	+ rows = fetch(months)
	104	+ if not rows:
	105	+ raise RuntimeError("[eurostat] no observations returned — refusing to write an empty file")
	106	+ fetched_at = datetime.now(UTC).isoformat()
	107	+ frame = (
	108	+ pl.DataFrame(rows)
	109	+ .with_columns(
	110	+ pl.lit("eurostat").alias("source"),
	111	+ pl.lit(fetched_at).alias("fetched_at"),
	112	+ )
	113	+ .unique(subset=["period", "airport_icao"], keep="last")
	114	+ .sort(["period", "airport_icao"])
	115	+ )
	116	+ target = Path(cfg.project_root) / "services" / "data" / "eurostat_airport_traffic.parquet"
	117	+ target.parent.mkdir(parents=True, exist_ok=True)
	118	+ frame.write_parquet(target, compression="zstd")
	119	+ logger.info(
	120	+ "[eurostat] %s rows · %s airports · %s..%s → %s",
	121	+ frame.height,
	122	+ frame["airport_icao"].n_unique(),
	123	+ frame["period"].min(),
	124	+ frame["period"].max(),
	125	+ target,
	126	+ )
	127	+ return target
	128	+
	129	+
	130	+def main() -> int:
	131	+ parser = argparse.ArgumentParser(description="Fetch Eurostat airport traffic")
	132	+ parser.parse_args()
	133	+ target = write_parquet()
	134	+ print(json.dumps({"file": str(target)}))
	135	+ return 0
	136	+
	137	+
	138	+if __name__ == "__main__":
	139	+ raise SystemExit(main())
0	140

▼scripts/publish_turso.py

+1-0

@@ -51,6 +51,7 @@ STATIC_TABLES = (
51	51	"gold_seasonal_trends",
52	52	"gold_fuel_price_series",
53	53	"gold_aircraft_class_mix",
	54	+ "gold_airport_official_traffic",
54	55	"site_stories",
55	56	"site_ops",
56	57	"site_lineage",
57	58

▼services/data/eurostat_airport_traffic.parquet

+0-0

Binary file not shown.