Airflow ETL Pipeline: Stock Data to SQLite End to End

You need stock prices in a database every morning. The API works, the transform works, but the pieces run by hand and the key sits in the code.

You'll build it as three tasks that run in order. Extract with a connection, flatten with pandas, load with a Hook.

We follow the Polygon-to-SQLite pattern: design table first, wire the graph, hide secrets properly. You'll have a runnable DAG by the end.

No keys in code. Ever.

Design First: Source, Destination, Transformation Plan

Write the tech spec before code: source endpoint and rate limit, destination table and key, transform rules for flattening nested JSON.

For Polygon stocks the grain is one row per symbol per date with an upsert key on (symbol, date). That choice makes reruns safe from day one.

Extract: Authenticated REST Call as a Task

Extract reads credentials from a Connection at runtime and returns raw JSON. No parsing here, just fetch and return.

Keep the task atomic: one symbol batch per run, with retries on 429 and 5xx. Pagination lives inside this task, not across tasks.

Give the call its own request timeout (30 seconds is sane) separate from retries, and declare retries on the decorator like @task(retries=3, retry_delay=timedelta(minutes=5)) so 429s and 5xx back off instead of dying. Write raw payloads to staging storage like S3 or /tmp keyed by ds and return the path, not the payload; XComs are for pointers, not 50,000-row dumps.

from airflow.decorators import dag, task
from airflow.hooks.base import BaseHook
from datetime import datetime
import requests

@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False, tags=["stocks"])
def stocks_etl():
    @task(retries=3)
    def extract_prices(data_interval_end=None):
        conn = BaseHook.get_connection("polygon_default")
        token = conn.password
        day = data_interval_end.strftime("%Y-%m-%d")
        resp = requests.get(
            f"https://api.polygon.io/v2/aggs/grouped/locale/us/market/stocks/{day}",
            params={"apiKey": token}, timeout=30,
        )
        resp.raise_for_status()
        return resp.json().get("results", [])

    @task
    def flatten_prices(rows: list):
        import pandas as pd
        df = pd.DataFrame(rows)
        df = df.rename(columns={"T": "symbol", "c": "close", "o": "open", "v": "volume"})
        return df[["symbol", "close", "open", "volume"]].to_dict("records")

    @task
    def load_prices(records: list):
        import sqlite3
        con = sqlite3.connect("/data/stocks.db")
        con.execute("CREATE TABLE IF NOT EXISTS daily_prices (symbol TEXT, date TEXT, open REAL, close REAL, volume INTEGER, PRIMARY KEY (symbol, date))")
        con.executemany("INSERT OR REPLACE INTO daily_prices VALUES (?,?,?,?,?)",
            [(r["symbol"], "2026-09-01", r["open"], r["close"], int(r["volume"])) for r in records])
        con.commit()
        return {"loaded": len(records)}

    load_prices(flatten_prices(extract_prices()))

stocks_etl()
# requests.get(url, timeout=30) — socket timeout is separate from task retries
# @task(retries=3, retry_delay=timedelta(minutes=5)) backs off 429/5xx
# requests.get(url, timeout=30) - socket timeout is separate from task retries
# return f"/tmp/sales_{ctx['ds']}.json" - pass paths via XCom, not payloads

Transform: JSON Flattening With Pandas

Flatten nested API payloads into a typed frame with explicit columns. Rename vendor fields to warehouse names in one place.

Coerce dtypes and reindex to expected columns so schema drift fails fast. A 50,000-row Polygon payload flattens in about 9 seconds.

Use @task(multiple_outputs=True) when transform returns a dict so each key becomes its own XCom instead of one opaque blob; downstream can then pull order_summary['total_order_value'] by name. Receive context explicitly with typed args like ds or ti instead of **kwargs soup; it's faster and self-documenting.

Load: Hook Plus Upsert

Loads must be idempotent: rerunning the same interval replaces the same rows. Use INSERT OR REPLACE on SQLite or ON CONFLICT on Postgres.

Create the table if missing, batch the executemany, and return the loaded count for lineage. Single-writer SQLite needs a Pool with one slot.

Wiring Dependencies and the Graph View

TaskFlow wires dependencies through function calls: load(flatten(extract())) draws a clean three-node graph. No manual bitshifts needed.

Check Graph view for extract, flatten, load in sequence. Each node should show one responsibility and typed inputs in the logs.

Mix styles freely: classic operators expose .output for TaskFlow calls, and decorated tasks accept .override(task_id=..., retries=...) for reuse across DAGs. When one task needs alien dependencies, reach for @task.virtualenv or @task.external_python instead of polluting the worker image; Docker or Kubernetes decorators isolate the truly exotic. Need context deep in a helper? get_current_context() fetches it without threading kwargs through every layer.

# Validate the ETL DAG end to end
airflow dags show stocks_etl
airflow dags test stocks_etl 2026-09-01
sqlite3 /data/stocks.db "SELECT COUNT(*) FROM daily_prices;"
# Check lineage in UI: Grid -> stocks_etl -> 2026-09-01

Secrets Through Connections, Never Code

Define polygon_default in UI or env as AIRFLOW_CONN_POLYGON_DEFAULT with the token in password. Dev uses env files, prod uses Vault.

Scan every PR with gitleaks and block merges on hits. Rotate by updating the backend; the DAG code never changes.