Airflow Tutorial Part 2: Loading to a Database

This is a hands-on tutorial that walks you through using Airflow, specifically loading data to a database.

Read Part 1 first to get your environment setup and extracting data: Airflow Tutorial #1: Your First Airflow Pipeline.

Subscribe to get notified when the next tutorial gets released:

Prerequisites:

• Completed Lab 1 first
• Docker Desktop is installed and running
• astro dev start has been run

What you’ll have when you’re done:

• A Postgres warehouse database running alongside Airflow
• A DAG (Direct Acyclic Graph) pipeline, rates_to_warehouse loading data to the Postgres table
• Proof of idempotency: ability to run it multiple times without getting duplicates

Let’s start.

Step 1 - Add the Postgres provider

This is how you install any integration in Airflow: name the provider, and Airflow rebuilds with its operators and hooks availables. -> Add providers

Open requirements.txt in your airflow-course project, and add this line:

apache-airflow-providers-postgres

Don’t restart yet - do step 2 first, then restart once.

Step 2 - Add a Postgres database

You’ll run a separate Postgres database for your data.

Note: Never load application data into Airflow’s own metadata database - that’s for Airflow’s internal state only.

In the top level of your project (same folder as requirements.txt), create a file named exactly:

docker-compose.override.yml

Put this in it:

services:
  warehouse:
    image: postgres:16
    environment:
      POSTGRES_USER: warehouse
      POSTGRES_PASSWORD: warehouse
      POSTGRES_DB: warehouse
    ports:
      - "5433:5432" # optional: lets you connect from your laptop on port 5433
    networks:
      - airflow # the network Astro's Airflow containers run on

Make sure the indentations stay the same.

The Astro CLI merges this file with its own Docker Compose config. -> Docker Compose overrides

Step 3 - Restart to apply both changes

astro dev restart

This rebuilds the image (installing the provider) and starts the warehouse container. It takes a couple of minutes. When it’s back, confirm the database container is running:

docker ps

You should see a container using the postgres:16 image with warehouse in its name.

Step 4 - Create a connection in Airflow

1. Go to the localhost port running your Airflow instance: http://localhost:8080 and login (admin / admin)
2. In the top nav, open Admin -> Connections.
3. Click + (Add Connection) and fill in:

Field	Value
Connection Id	warehouse_db
Connection Type	Postgres
Host	warehouse
Database (Schema)	warehouse
Login	warehouse
Password	warehouse
Port	5432

4. Click Save.

Step 5 - Create the Variable in Airflow

A variable is a named value that Airflow stores for you and hands to your DAGs at runtime, kept separately from your code.

Why use it: To avoid hard-coding values inside the code, so if a config changes (file paths, a feature flag, new list of currencies etc.), you don’t have to edit the code and redeploy. You can also use this to separate dev vs. prod environments.

Note: Variables are for configuration, not secrets. They’re readable in plain text in the UI, so passwords and tokens belong in Connections (Step 4), which are encrypted.

In this lab, the currencies to fetch are config, so they go in a Variable. Create it:

1. In the top nav, open Admin → Variables.
2. Click + and enter:
   • Key: exchange_rate_config
   • Value: {"base": "USD", "symbols": "EUR,GBP,JPY,CAD,AUD"}
3. Click Save.

You’ll read this in the DAG with Variable.get(..., deserialize_json=True), so changing the currencies later needs no code change. → Variables

Step 6 - Create the DAG

In dags/, create a file named exactly rates_to_warehouse.py.

Highly recommended: We highly recommend typing this out by hand so you can get a feel for what functions, layout, operators, hooks, etc. are being used.

"""
Lab 2 — The Reliable Loader
Extract USD exchange rates -> load into Postgres idempotently.
"""
from __future__ import annotations

import json
import urllib.parse
import urllib.request
from datetime import timedelta

import pendulum

from airflow.sdk import Variable, dag, task
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook

CONN_ID = "warehouse_db"
TABLE = "exchange_rates"


@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
    catchup=False,
    default_args={
        "retries": 2,                       # retry a failed task twice...
        "retry_delay": timedelta(minutes=1),  # ...waiting 1 min between attempts
    },
    tags=["lab-2", "course"],
)
def rates_to_warehouse():

    # An OPERATOR: create the table if it doesn't exist. Safe to run every time.
    create_table = SQLExecuteQueryOperator(
        task_id="create_table",
        conn_id=CONN_ID,
        sql=f"""
            CREATE TABLE IF NOT EXISTS {TABLE} (
                rate_date DATE NOT NULL,
                base      TEXT NOT NULL,
                target    TEXT NOT NULL,
                rate      NUMERIC NOT NULL
            );
        """,
    )

    @task
    def extract() -> dict:
        # Read config from a Variable instead of hard-coding it.
        config = Variable.get("exchange_rate_config", deserialize_json=True)
        query = urllib.parse.urlencode(
            {"base": config["base"], "symbols": config["symbols"]}
        )
        url = f"https://api.frankfurter.app/latest?{query}"

        # frankfurter.app's Cloudflare front-end 403s the default

		# "Python-urllib/x.x" User-Agent, so send a browser-like one.

		request = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})

        with urllib.request.urlopen(url, timeout=30) as response:
            payload = json.loads(response.read().decode())
        print(f"Fetched {len(payload['rates'])} rates dated {payload['date']}")
        return payload

    @task
    def transform(payload: dict) -> list[dict]:
        return [
            {
                "rate_date": payload["date"],
                "base": payload["base"],
                "target": currency,
                "rate": rate,
            }
            for currency, rate in payload["rates"].items()
        ]

    @task
    def load(rows: list[dict]) -> None:
        # A HOOK: gives direct control over the database.
        hook = PostgresHook(postgres_conn_id=CONN_ID)
        rate_date = rows[0]["rate_date"]

        # IDEMPOTENCY: delete this date's rows, then insert. Re-running is safe.
        hook.run(f"DELETE FROM {TABLE} WHERE rate_date = %s", parameters=(rate_date,))
        hook.insert_rows(
            table=TABLE,
            rows=[(r["rate_date"], r["base"], r["target"], r["rate"]) for r in rows],
            target_fields=["rate_date", "base", "target", "rate"],
        )
        print(f"Loaded {len(rows)} rows for {rate_date}")

    @task
    def count_rows() -> None:
        hook = PostgresHook(postgres_conn_id=CONN_ID)
        total = hook.get_first(f"SELECT COUNT(*) FROM {TABLE}")[0]
        print(f"Table {TABLE} now has {total} total rows")

    payload = extract()
    rows = transform(payload)
    loaded = load(rows)

    # load waits for BOTH the table (explicit) and the data (via rows).
    create_table >> loaded >> count_rows()


rates_to_warehouse()

Save. If it doesn’t appear or errors, run astro dev parse to see the problem.

Step 7 - Run it

1. At http://localhost:8080, find rates_to_warehouse and unpause it.
2. Open it and click ▶ Trigger.
3. Watch create_table → extract → transform → load → count_rows go green.

Step 8 — Check your work

In the UI:

• Click the count_rows task → Logs. You should see Table exchange_rates now has 5 total rows.
• Click the load task → Logs → Loaded 5 rows for YYYY-MM-DD.

Prove idempotency: Trigger the DAG 4 more times. Then check the count_rows log again: it will still say 5 total rows, not 25. The DELETE-then-insert means each run replaces the date’s data instead of piling on duplicates.

Look at the data directly inside the Postgres db:

Find the warehouse container name, then open a SQL shell inside it:

bash

docker ps
# copy the container name that uses postgres:16 (has "warehouse" in it), then:
docker exec -it <container-name> psql -U warehouse -d warehouse

At the warehouse=# prompt:

sql

SELECT * FROM exchange_rates;
\q

Five rows, one per currency. → PostgresHook methods

You’re done!

At this point now, here’s what you should have:

• warehouse Postgres container is running (docker ps).
• warehouse_db connection and exchange_rate_config variable exist in the UI.
• A run completes with all five tasks green.
• After triggering 5 times, count_rows still reports 5 rows.

Summary of Learnings:

• Connection — a saved login for an outside system, so your code just says “use warehouse_db” instead of holding the password. → Connections
• Hook — the tool that talks to that system for you (e.g. PostgresHook). It uses the Connection and gives you methods like run() and insert_rows(). → PostgresHook
• Operator — a ready-made task. SQLExecuteQueryOperator runs SQL for you. → Operators
• Provider package — an add-on you install to get operators and hooks for one system. Postgres comes from apache-airflow-providers-postgres. → Provider registry
• Variable — a setting you store in Airflow instead of typing it in your code. For config, not secrets. → Variables
• Retries — if a task fails, Airflow tries it again. Handy when an API blips. → Tasks / retries
• Idempotency — run it once or five times, you get the same result — no duplicates. → Best practices

Did you like this?

This is just one of many hands-on tutorials for data engineering and data work. If you liked this tutorial, subscribe to get notified when the next tutorial comes out.

What’s next — Lab 3 preview

Your tasks already hand data to each other (extract → transform → load). Next lab, “Passing the Baton,” you’ll look under the hood at XComs — how that hand-off actually works, its size limits, and the critical rule that keeps pipelines fast.

Shut down

As always, don’t forget to shut down your Airflow instance when you’re done.

astro dev stop

Your data and connection persist. astro dev start brings everything back. (Only astro dev kill wipes the databases.)