Data Pipeline — Technical Overview

A lightweight, four-layer pipeline: an ingestion script pulls from the Duffel API, a relational store normalizes the data, a prediction service trains models nightly, and a thin API layer feeds the frontend.

A single public API source for flight pricing data.

Idempotent upserts with composite unique keys — re-running the script never creates duplicates.

import httpx, os, pandas as pd
from datetime import date, timedelta

def fetch_flight_prices(api_key, route, depart_date):
    """Pull flight prices from the flight search API."""
    base_url = os.environ["FLIGHT_API_URL"]
    resp = httpx.get(
        base_url + "/v1/search",
        params={"from_iata_code": route.origin, "to_iata_code": route.dest},
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=30
    )
    resp.raise_for_status()
    results = resp.json()["results"]
    rows = []
    for r in results:
        for seg in r["segments"]:
            if seg["type"] == "flight" and seg.get("price"):
                rows.append({
                    "route_code": f"{route.origin}-{route.dest}",
                    "observation_date": date.today(),
                    "departure_date": depart_date,
                    "price_cents": seg["price"]["value"],
                    "currency": seg["price"]["currency"],
                    "category": classify_route(route),
                    "source": "duffel",
                })
    return pd.DataFrame(rows)

def upsert_prices(df, conn):
    """Idempotent upsert — UNIQUE(route_code, observation_date, departure_date, source)."""
    df.to_sql("ft_price_observations", conn, if_exists="append", index=False,
              method="multi", chunksize=50)
    # ON CONFLICT DO UPDATE at DB level

A relational store normalized for time-series queries and fast filtering by route / category / date.

Two models trained nightly at 02:00 against the accumulated price history. Forecasts cover a 3–90 day horizon before departure.

Thin endpoints — each is a single query wrapped in JSON. No business logic; all computation lives in the prediction service or the frontend.

# FastAPI endpoint (simplified)
@app.get("/api/free/trends")
def get_trends(currency: str = "EUR"):
    """Return price trend index per category (chain-indexed, start=100)."""
    conn = get_conn()
    cur = conn.cursor()
    cur.execute("""
        SELECT category, observation_date,
               percentile_cont(0.5) WITHIN GROUP (ORDER BY price) AS median
        FROM ft_price_observations
        WHERE observation_date >= now() - interval '30 days'
        GROUP BY category, observation_date
        ORDER BY category, observation_date
    """)
    rows = cur.fetchall()
    conn.close()
    return {"ok": True, "trends": build_chain_index(rows, currency)}

Vanilla JavaScript with Chart.js 4.x. No build step, no framework — each section is a standalone module that fetches from the API and renders into its container.

// index.html — data loading + chart rendering
var API_BASE = window.location.origin + '/api/flight-tracker';
var ftLang = localStorage.getItem("seb-lang") || "en";

function loadData() {
  var p1 = fetch(API_BASE + '?view=free¤cy=' + currentCurrency);
  var p2 = fetch(API_BASE + '?view=forecast¤cy=' + currentCurrency);
  Promise.all([p1, p2]).then(function(results) {
    allData = results[0];
    forecastData = results[1];
    render();
  });
}

function renderTrendChart(trends) {
  var ctx = document.getElementById('trend-chart');
  new Chart(ctx, {
    type: 'line',
    data: {
      labels: trends.dates.map(d => d.slice(5)),
      datasets: trends.series.map((s, i) => ({
        label: s.name,
        data: s.index,
        borderColor: COLORS[i],
        tension: 0.3,
        pointRadius: 2
      }))
    },
    options: { responsive: true }
  });
}

What happens between the raw API response and the data store.

All publicly exposed data is anonymized. No passenger data, no airline identifiers beyond IATA codes, no individual booking records. Prices are aggregated per route and date — never per traveller.