Ingestion Pipelines with dlt

5.1 Why a loading tool? Your first dlt pipeline (40 min)

Concept. Yesterday's client works, but a production loader needs more: create tables with the right types, handle new columns, remember what was already loaded, never load an order twice, and record every load. dlt (data load tool) does all of this. You write a Python function that yields records; dlt turns them into tables.

Make sure the API is running in your second terminal, then create day05/dlt_hello.py:

# day05/dlt_hello.py
from pathlib import Path

import dlt

# 1. Any list of dicts (or a generator of them) can be loaded
rates = [
    {"week": "2026-03-02", "segment": "loyalty", "orders_per_customer": 0.52},
    {"week": "2026-03-02", "segment": "regular", "orders_per_customer": 0.21},
]

# 2. A pipeline = a name + a destination + a dataset (schema) to write into
pipeline = dlt.pipeline(
    pipeline_name="hello",
    destination=dlt.destinations.duckdb(str(Path("data/northmart.duckdb").resolve())),
    pipelines_dir=".dlt_pipelines",   # keep pipeline state inside the project
    dataset_name="practice",
)

# 3. run() infers the schema, creates the table and loads the rows
info = pipeline.run(rates, table_name="weekly_rates")
print(info)

# 4. Read it back
with pipeline.sql_client() as client:
    print(client.execute_sql("SELECT * FROM weekly_rates"))

python day05/dlt_hello.py

Pipeline hello load step finished in 0.23 seconds
1 load package(s) were loaded to destination duckdb and into dataset practice
The duckdb destination used duckdb:////workspaces/causal-decision-lab/data/northmart.duckdb location to store data
Load package 1793486741.6855483 is LOADED and contains no failed jobs
[('2026-03-02', 'loyalty', 0.52, '1793486741.6855483', 'cKfwif9Og0Edog'), ('2026-03-02', 'regular', 0.21, ...)]

dlt created the schema practice and the table weekly_rates, and added two columns of its own: _dlt_load_id (which load wrote the row) and _dlt_id (a unique row ID). Every row can be traced back to the load that wrote it.

Two settings in the pipeline deserve a sentence each:

• An absolute database path (Path(...).resolve()). dlt remembers each pipeline between runs, including the folder it was first run from. A relative path can then point somewhere unexpected if you run the script from a different folder.
• pipelines_dir=".dlt_pipelines" keeps that memory inside your project (and out of Git), instead of a hidden system folder shared with other projects.

Now run the same script a second time. The final line shows four rows. By default dlt appends: running a load twice duplicates the data. Section 5.3 explains how to choose.

5.2 Resources: from API pages to tables (50 min)

Concept. A dlt resource is a generator decorated with @dlt.resource. Yesterday's pagination loop becomes a resource almost unchanged: instead of collecting rows into a list, it yields each page.

Create day05/orders_resource.py:

# day05/orders_resource.py
from pathlib import Path

import dlt
import requests

BASE = "http://127.0.0.1:8000"
HEADERS = {"X-API-Key": "northmart-demo-key"}


@dlt.resource(name="orders", write_disposition="replace")
def orders():
    """A resource is a generator: it yields pages, dlt does the rest."""
    page = 1
    while page is not None:
        r = requests.get(f"{BASE}/orders", headers=HEADERS,
                         params={"page": page, "page_size": 1000}, timeout=30)
        r.raise_for_status()
        body = r.json()
        yield body["data"]
        page = body["next_page"]


pipeline = dlt.pipeline(
    pipeline_name="orders_demo",
    destination=dlt.destinations.duckdb(str(Path("data/northmart.duckdb").resolve())),
    pipelines_dir=".dlt_pipelines",   # keep pipeline state inside the project
    dataset_name="practice",
)
print(pipeline.run(orders()))

with pipeline.sql_client() as client:
    print(client.execute_sql("SELECT COUNT(*), COUNT(DISTINCT order_id) FROM orders"))
    cols = client.execute_sql("""
        SELECT column_name, data_type FROM information_schema.columns
        WHERE table_schema = 'practice' AND table_name = 'orders' ORDER BY ordinal_position
    """)
    for name, dtype in cols:
        print(f"  {name:<22} {dtype}")

[(18542, 18359)]
  order_id               VARCHAR
  customer_id            VARCHAR
  order_date             VARCHAR
  status                 VARCHAR
  updated_at             TIMESTAMP WITH TIME ZONE
  line__product_id       VARCHAR
  line__quantity         BIGINT
  line__unit_price       DOUBLE
  line__discount_pct     BIGINT
  _dlt_load_id           VARCHAR
  _dlt_id                VARCHAR

Three things happened without you asking:

1. Flattening. The nested line object became line__product_id, line__quantity… (double underscore = one level of nesting).
2. Type inference. updated_at was recognised as a timestamp. order_date stayed text: dlt is cautious with dates, so we cast it ourselves in dbt.
3. Nothing was deduplicated. 18,542 rows but only 18,359 distinct order_ids: the API serves the same 183 duplicate rows as the CSV.

5.3 Write dispositions: replace, append, merge (50 min)

Concept. The write disposition decides what happens to existing rows when you load again:

merge needs a primary key: the column(s) that identify a record. With primary_key="order_id", loading the same order twice keeps one row. That fixes the duplicate problem at the source rather than downstream.

Try it. Load the same two rows twice with append, and twice with replace, into separate datasets. How many rows does each table hold?

from pathlib import Path

import dlt

rows = [{"segment": "loyalty", "orders": 0.52}, {"segment": "regular", "orders": 0.21}]

for disposition in ["append", "replace"]:
    pipeline = dlt.pipeline(
        pipeline_name=f"demo_{disposition}",
        destination=dlt.destinations.duckdb(str(Path("data/northmart.duckdb").resolve())),
        pipelines_dir=".dlt_pipelines",
        dataset_name=f"practice_{disposition}",
    )
    for _ in range(2):                      # run the same load twice
        pipeline.run(rows, table_name="rates", write_disposition=disposition)
    with pipeline.sql_client() as client:
        n = client.execute_sql("SELECT COUNT(*) FROM rates")[0][0]
    print(f"{disposition:<8} after two runs: {n} rows")

append   after two runs: 4 rows
replace  after two runs: 2 rows

5.4 Incremental loading as time passes (45 min)

Concept. NorthMart's orders grow every day. Reloading 16 weeks of history every morning wastes time and, for big companies, money. Incremental loading asks the API only for what changed since the last run. dlt stores the cursor (the largest updated_at seen) in the pipeline's state and hands it back next time.

The API's clock lets us replay time. Before the email: set the clock to Saturday 28 February 2026, the last day of week 8.

Create day05/incremental_orders.py:

# day05/incremental_orders.py
"""Load only what is new, and never load the same order twice."""
from pathlib import Path

import dlt
import requests

BASE = "http://127.0.0.1:8000"
HEADERS = {"X-API-Key": "northmart-demo-key"}


@dlt.resource(name="orders", primary_key="order_id", write_disposition="merge")
def orders(updated_at=dlt.sources.incremental("updated_at", initial_value="2026-01-01T00:00:00Z")):
    # dlt remembers the largest updated_at it has seen and gives it back to us next run.
    # Read it ONCE: last_value keeps moving while pages arrive during this run.
    since = updated_at.start_value
    print(f"  asking the API for orders updated after {since}")
    page = 1
    while page is not None:
        r = requests.get(f"{BASE}/orders", headers=HEADERS, timeout=30,
                         params={"page": page, "page_size": 1000, "updated_since": since})
        r.raise_for_status()
        body = r.json()
        yield body["data"]
        page = body["next_page"]


pipeline = dlt.pipeline(
    pipeline_name="incremental_demo",
    destination=dlt.destinations.duckdb(str(Path("data/northmart.duckdb").resolve())),
    pipelines_dir=".dlt_pipelines",
    dataset_name="practice_incremental",
)
pipeline.run(orders())

with pipeline.sql_client() as client:
    n, n_ids, last = client.execute_sql(
        "SELECT COUNT(*), COUNT(DISTINCT order_id), MAX(order_date) FROM orders")[0]
    print(f"orders table: {n:,} rows, {n_ids:,} distinct order_ids, latest order {last}")

Run it three times, moving the clock as a real company would experience it:

python api/clock.py 2026-02-28
python day05/incremental_orders.py      # first run: everything up to 28 Feb
python day05/incremental_orders.py      # same day again: nothing new
python api/clock.py 2026-04-26
python day05/incremental_orders.py      # eight weeks later: only the new orders

NorthMart API 'today' is now 2026-02-28
  asking the API for orders updated after 2026-01-01T00:00:00Z
orders table: 9,287 rows, 9,287 distinct order_ids, latest order 2026-02-28
  asking the API for orders updated after 2026-02-28T23:00:00Z
orders table: 9,287 rows, 9,287 distinct order_ids, latest order 2026-02-28
NorthMart API 'today' is now 2026-04-26
  asking the API for orders updated after 2026-02-28T23:00:00Z
orders table: 18,359 rows, 18,359 distinct order_ids, latest order 2026-04-26

The final table has exactly one row per order (18,359), and the duplicates are gone because of merge.

A bug worth knowing about. The code reads updated_at.start_value once, before the loop. last_value changes during the run as rows arrive. If you pass last_value to every page request, later pages ask for "orders after the newest one seen so far" and silently skip older orders on those pages. In testing, that version loaded 2,978 orders instead of 9,287 and raised no error. Incremental code must be checked against a full reload at least once.

Late data. Our API never changes an order after its date. Real systems do (refunds, corrections). A common safeguard is a lookback: ask for everything updated in the last few days, and let merge absorb the overlap.

5.5 Day 5 mini-project: the NorthMart source (70 min)

Goal. Group all four endpoints into one dlt source that loads into the schema raw, with settings in .dlt/config.toml and the key in .dlt/secrets.toml:

Step 1 — configuration and secrets. Create the folder .dlt/ and two files:

# .dlt/config.toml  (safe to commit: no secrets here)
[sources.northmart]
base_url = "http://127.0.0.1:8000"

# .dlt/secrets.toml  (NEVER commit this file: it is listed in .gitignore)
[sources.northmart]
api_key = "northmart-demo-key"

dlt fills any argument declared as dlt.config.value or dlt.secrets.value from these files, matched by name and section. In production you'd set the environment variable SOURCES__NORTHMART__API_KEY instead; dlt checks it automatically.

Step 2 — the source. Create day05/northmart_pipeline.py:

# day05/northmart_pipeline.py
"""Mini-project: one dlt source for the whole NorthMart API, loaded into the `raw` schema."""
from pathlib import Path

import dlt
import requests

DB = str(Path("data/northmart.duckdb").resolve())


def get_pages(base_url: str, api_key: str, endpoint: str, **params):
    page = 1
    while page is not None:
        r = requests.get(f"{base_url}/{endpoint}", headers={"X-API-Key": api_key}, timeout=30,
                         params={"page": page, "page_size": 1000, **params})
        r.raise_for_status()
        body = r.json()
        yield body["data"]
        page = body.get("next_page")


def get_pages_once(base_url: str, api_key: str, endpoint: str) -> list[dict]:
    r = requests.get(f"{base_url}/{endpoint}", headers={"X-API-Key": api_key}, timeout=30)
    r.raise_for_status()
    return r.json()["data"]


@dlt.source(name="northmart")
def northmart_source(base_url: str = dlt.config.value, api_key: str = dlt.secrets.value):
    """base_url comes from .dlt/config.toml, api_key from .dlt/secrets.toml (or env vars)."""

    @dlt.resource(primary_key="customer_id", write_disposition="replace")
    def customers():
        yield from get_pages(base_url, api_key, "customers")

    @dlt.resource(primary_key="product_id", write_disposition="replace")
    def products():
        yield get_pages_once(base_url, api_key, "products")

    @dlt.resource(primary_key="customer_id", write_disposition="replace")
    def campaign_recipients():
        body = get_pages_once(base_url, api_key, "campaigns/SPRING26/recipients")
        yield [{**row, "campaign_id": "SPRING26"} for row in body]

    @dlt.resource(primary_key="order_id", write_disposition="merge")
    def orders(updated_at=dlt.sources.incremental("updated_at", initial_value="2026-01-01T00:00:00Z")):
        since = updated_at.start_value
        yield from get_pages(base_url, api_key, "orders", updated_since=since)

    return customers, products, campaign_recipients, orders


def main() -> None:
    pipeline = dlt.pipeline(
        pipeline_name="northmart",
        destination=dlt.destinations.duckdb(DB),
        pipelines_dir=".dlt_pipelines",
        dataset_name="raw",
    )
    pipeline.run(northmart_source())

    with pipeline.sql_client() as client:
        existing = {r[0] for r in client.execute_sql(
            "SELECT table_name FROM information_schema.tables WHERE table_schema = 'raw'")}
        for table in ["customers", "products", "campaign_recipients", "orders"]:
            if table not in existing:      # dlt only creates a table once it has received rows
                print(f"raw.{table:<20}  (no rows yet)")
                continue
            n = client.execute_sql(f"SELECT COUNT(*) FROM {table}")[0][0]
            print(f"raw.{table:<20} {n:>7,} rows")


if __name__ == "__main__":
    main()

Step 3 — replay the timeline.

python api/clock.py 2026-02-28
python day05/northmart_pipeline.py
python api/clock.py 2026-04-26
python day05/northmart_pipeline.py

NorthMart API 'today' is now 2026-02-28
raw.customers              4,000 rows
raw.products                  40 rows
raw.campaign_recipients   (no rows yet)
raw.orders                 9,287 rows
NorthMart API 'today' is now 2026-04-26
raw.customers              4,000 rows
raw.products                  40 rows
raw.campaign_recipients    1,279 rows
raw.orders                18,359 rows

On 28 February the email hasn't been sent, so there are no recipients, and dlt doesn't create a table until it receives rows. The script checks for that instead of crashing.

Try it. dlt records every load. Query raw._dlt_loads, then count orders by _dlt_load_id. Can you see the two loads?

import duckdb

con = duckdb.connect("data/northmart.duckdb", read_only=True)
print(con.sql("""
    SELECT load_id, schema_name, status, inserted_at
    FROM raw._dlt_loads
    ORDER BY inserted_at
"""))
print(con.sql("""
    SELECT _dlt_load_id, COUNT(*) AS orders_loaded
    FROM raw.orders
    GROUP BY _dlt_load_id
    ORDER BY _dlt_load_id
"""))

│    _dlt_load_id    │ orders_loaded │
├────────────────────┼───────────────┤
│ 1793487005.3942654 │          9287 │
│ 1793487010.2876098 │          9072 │

5.6 Save your work with Git (15 min)

git add .
git status          # .dlt/secrets.toml and .dlt_pipelines/ must NOT appear
git commit -m "Day 5: dlt source with incremental, deduplicated orders"
git push

If secrets.toml appears in git status, stop and fix .gitignore before committing.

Day 5 checklist

• I can write a dlt resource from a paginated API and load it into DuckDB
• I can choose between replace, append and merge, and explain why orders need merge
• I can load incrementally with a cursor and explain the start_value pitfall
• I keep settings in config.toml and keys in secrets.toml, never in Git
• I can trace any row back to the load that wrote it