Airflow ETL Pipeline: Stock Data to SQLite, End to End

Every data team builds this eventually: pull rows from an API, clean them, stick them in a database. The shape of the DAG decides whether that pipeline is a joy to run or a 2 AM pager rotation.

We're building a stock-price ETL: Polygon REST API into a SQLite database, using the TaskFlow API, pandas for the transform, and SqliteHook for the load. Three tasks, each atomic, each retryable, each debuggable in isolation. That's the whole lesson in one line.

The traps are real and familiar: API keys hardcoded in the repo, 10 MB payloads pushed through XCom, loads that duplicate rows on every rerun. We'll hit each one and fix it properly. By the end you'll have a pipeline you'd actually put in production — and the muscle memory to build the next one faster.

1. Design First: Source, Destination, Transformation Plan

Before any code, write the contract. The Polygon API gives previous-day aggregates per ticker; the destination is a SQLite table; the transform turns nested JSON into flat rows. Writing this down first kills half the bugs before they exist.

This is the tech spec we'll build against. Notice what it pins down: the endpoint, the auth method, the target schema, the dedupe key. If a requirement changes, you change the spec, not the DAG.

2. Extract: Authenticated REST Call as a Task

Extract is one task, one responsibility: hit the API, return the parsed JSON. Keep it dumb. No cleaning here, no filtering — the transform owns that.

Two production habits make this task survive: the API key comes from the connection (never from a constant), and the request has a timeout. An API that hangs without one steals a worker slot for the full task timeout.

Know the endpoint family. /v2/aggs/ticker/{ticker}/prev returns the previous trading session; its sibling /v1/open-close/{ticker}/{date}?adjusted=true returns one specific day's open-close bars, splits-adjusted when adjusted=true — the right call when "previous day" isn't the day you need.

The task returns the payload, and TaskFlow pushes it through XCom for the transform. Keep the payload small; for heavy responses, write to object storage and pass the URI instead.

import requests
from airflow.decorators import task
from airflow.hooks.base import BaseHook


@task

def extract_stock_data(tickers: list[str]) -> dict:
    conn = BaseHook.get_connection("market_data_api")
    token = conn.extra_dejson["token"]
    base_url = conn.host or "https://api.polygon.io"

    payloads = {}
    for ticker in tickers:
        resp = requests.get(
            f"{base_url}/v2/aggs/ticker/{ticker}/prev",
            headers={"Authorization": f"Bearer {token}"},
            timeout=30,
        )
        resp.raise_for_status()
        payloads[ticker] = resp.json()
    return payloads

3. Transform: Flattening JSON with pandas

Polygon returns nested results arrays with short field names: o, h, l, c, v, T. The transform expands them into readable columns and coerces types — this is where the schema contract from section 1 comes alive.

Two rules keep transform tasks honest: every field the load expects must exist (assert it), and the task returns a compact payload, typically rows as a list of dicts or a CSV path. Don't return a DataFrame through XCom; it serializes badly.

A transform task that produces nothing silently is the worst kind of failure. End it with a row-count check.

One Polygon quirk to expect: on market-closed days — weekends and holidays — the response body is sparse and the result keys are missing entirely. The row-count guard turns that into a loud, deliberate failure; decide per pipeline whether a closed day is an error or a legitimate empty batch, and make the transform's contract say which.

import pandas as pd
from airflow.decorators import task


@task
def transform_stock_payload(payloads: dict) -> list[dict]:
    frames = []
    for ticker, body in payloads.items():
        frame = pd.DataFrame(body.get("results", []))
        frame["ticker"] = ticker
        frame = frame.rename(columns={
            "T": "date", "o": "open", "h": "high",
            "l": "low", "c": "close", "v": "volume",
        })
        frames.append(frame)

    combined = pd.concat(frames, ignore_index=True)
    combined["date"] = pd.to_datetime(combined["date"]).dt.date
    combined = combined[["ticker", "date", "open", "high", "low", "close", "volume"]]

    rows = combined.to_dict("records")
    if not rows:
        raise ValueError("Transform produced zero rows")
    return rows

4. Load: SqliteHook + SQLAlchemy + Upsert

The load task opens the database through SqliteHook, builds a SQLAlchemy engine from the connection URI, and upserts on the (ticker, date) key. Idempotent by construction: rerun it a hundred times, the table holds the same rows.

Why a hook instead of a raw sqlite3 call? The hook owns the connection config — the same DAG runs unchanged across dev and prod because the database path lives in the connection. The engine gives you to_sql for bulk inserts and parameterized upsert for control.

One trap: SQLite is a file, and "where is the file" is a connection detail. Two connections pointing at two files is a classic "green run, empty table" bug.

If the target ever lacks a natural unique key, the classic idempotency pattern is delete-then-insert keyed on run date: delete the run's rows first, then insert — replays produce identical tables without needing ON CONFLICT.

from airflow.decorators import task
from airflow.providers.sqlite.hooks.sqlite import SqliteHook
from sqlalchemy import text


@task
def load_stock_prices(rows: list[dict]) -> int:
    hook = SqliteHook(sqlite_conn_id="market_database_conn")
    engine = hook.get_sqlalchemy_engine()

    with engine.begin() as conn:
        for row in rows:
            conn.execute(
                text("""
                    INSERT INTO stock_prices
                        (ticker, date, open, high, low, close, volume)
                    VALUES (:ticker, :date, :open, :high, :low, :close, :volume)
                    ON CONFLICT (ticker, date) DO UPDATE SET
                        open = excluded.open,
                        high = excluded.high,
                        low = excluded.low,
                        close = excluded.close,
                        volume = excluded.volume
                """),
                row,
            )
    return len(rows)

5. The Full DAG: Wiring It Together

Here's the complete file, from imports to schedule. Three tasks, explicit dependencies, sensible defaults: retries on transient API blips, catchup off for a pipeline that only needs today, and tags so the team finds it in the UI.

Note what's absent: no API key, no file path, no credentials. Everything environment-specific sits in connections. The DAG is deployable to dev or prod unchanged.

This is the shape to replicate for every ETL you build: extract, transform, load, wired left to right, each task independently retryable.

from datetime import datetime, timedelta

import pandas as pd
import requests
from airflow.decorators import dag, task
from airflow.hooks.base import BaseHook
from airflow.providers.sqlite.hooks.sqlite import SqliteHook
from sqlalchemy import text

TICKERS = ["AAPL", "MSFT", "GOOGL", "NVDA"]


@task
def extract_stock_data(tickers: list[str]) -> dict:
    conn = BaseHook.get_connection("market_data_api")
    token = conn.extra_dejson["token"]
    base_url = conn.host or "https://api.polygon.io"

    payloads = {}
    for ticker in tickers:
        resp = requests.get(
            f"{base_url}/v2/aggs/ticker/{ticker}/prev",
            headers={"Authorization": f"Bearer {token}"},
            timeout=30,
        )
        resp.raise_for_status()
        payloads[ticker] = resp.json()
    return payloads


@task
def transform_stock_payload(payloads: dict) -> list[dict]:
    frames = []
    for ticker, body in payloads.items():
        frame = pd.DataFrame(body.get("results", []))
        frame["ticker"] = ticker
        frame = frame.rename(columns={
            "T": "date", "o": "open", "h": "high",
            "l": "low", "c": "close", "v": "volume",
        })
        frames.append(frame)

    combined = pd.concat(frames, ignore_index=True)
    combined["date"] = pd.to_datetime(combined["date"]).dt.date
    combined = combined[["ticker", "date", "open", "high", "low", "close", "volume"]]

    rows = combined.to_dict("records")
    if not rows:
        raise ValueError("Transform produced zero rows")
    return rows


@task
def load_stock_prices(rows: list[dict]) -> int:
    hook = SqliteHook(sqlite_conn_id="market_database_conn")
    engine = hook.get_sqlalchemy_engine()

    with engine.begin() as conn:
        for row in rows:
            conn.execute(
                text("""
                    INSERT INTO stock_prices
                        (ticker, date, open, high, low, close, volume)
                    VALUES (:ticker, :date, :open, :high, :low, :close, :volume)
                    ON CONFLICT (ticker, date) DO UPDATE SET
                        open = excluded.open,
                        high = excluded.high,
                        low = excluded.low,
                        close = excluded.close,
                        volume = excluded.volume
                """),
                row,
            )
    return len(rows)


@dag(
    schedule="@daily",
    start_date=datetime(2026, 7, 1),
    catchup=False,
    tags=["market", "etl"],
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)
def market_etl():
    extract_stock_data(TICKERS) >> transform_stock_payload() >> load_stock_prices()


market_etl()

6. Secrets Through Connections, Never Code

The incident that opened this article had one cause: a key in code. The fix is structural, not stylistic. Credentials live in Airflow connections; the DAG references connection ids; the secrets backend (Vault, AWS Secrets Manager) holds the values.

Rotation becomes a config change instead of a deploy: update the secret, and every DAG picks it up on the next run. No redeploy, no rollback dance, no git history archaeology.

Add the scan to CI, too. The key already in the repo's history is the one you can't delete by deleting the line.

7. Running, Verifying, and Extending

To verify without waiting for the schedule: airflow dags test market_etl runs the whole DAG once, in the scheduler's context, and streams the logs. Then check the database with a quick count and a duplicate check — the two queries every ETL review should run.

``bash airflow dags test market_etl 2026-08-01 sqlite3 /opt/airflow/data/market.db "SELECT COUNT() FROM stock_prices;" sqlite3 /opt/airflow/data/market.db \ "SELECT ticker, date, COUNT() FROM stock_prices GROUP BY 1,2 HAVING COUNT(*) > 1;" ``

Extending the pipeline is mechanical now: add a ticker to TICKERS, or map over tickers with expand() when the list grows. Add a quality gate task after load — row count matches the API response count — and you have a production pipeline.

A stronger gate than a row count is a schema validation task between transform and load: pandera or Great Expectations checks required fields, types, and value ranges, and fails before a single row touches the table — the same checks that used to live as buried asserts become a visible, testable task that can page someone.

The 40-second runtime vs the 15-minute cron script isn't the win. The win is failure isolation: when the API is down, only extract fails, and it retries on its own.