Configuration, Secrets and the macro_ingest Package

23.1 Config vs Secrets (20 min)

Your code so far has values hard-coded in it: series lists, start dates, file names. That causes three problems:

1. Changing what you fetch means editing code.
2. Different environments (your laptop, a server, a colleague's machine) need different settings.
3. Secrets (API keys, passwords) end up in Git.

dlt separates two kinds of settings:

Create the folder and both files:

mkdir -p .dlt
touch .dlt/config.toml .dlt/secrets.toml
git check-ignore .dlt/secrets.toml && echo "✓ secrets.toml is ignored"

If the last command prints nothing, fix .gitignore (Day 21.1) before continuing.

TOML in two minutes

TOML is a simple, readable settings format:

# comments start with #
[runtime]                    # a section
log_level = "WARNING"        # text in quotes

[sources.boc]                # a nested section: sources → boc
series_ids = ["V39079", "FXCADUSD"]   # a list
start_date = "2006-01-01"

[sources.statcan]
vectors = [41690973, 2062815]         # numbers without quotes
latest_n = 250

Sections map naturally to Python dictionaries: sources → boc → series_ids.

23.2 Injection: dlt.config.value and dlt.secrets.value (30 min)

Instead of passing arguments yourself, you can ask dlt to inject them from configuration.

Example 1 — Required configuration

Put this in .dlt/config.toml:

[sources.boc]
series_ids = ["V39079", "FXCADUSD", "BD.CDN.2YR.DQ.YLD", "BD.CDN.10YR.DQ.YLD"]
start_date = "2006-01-01"

Create day23/config_demo.py:

import dlt


@dlt.source(name="boc", section="boc")
def boc_config_demo(series_ids: list = dlt.config.value,
                    start_date: str = dlt.config.value):

    @dlt.resource(name="boc_settings_echo", write_disposition="replace")
    def echo():
        for sid in series_ids:
            yield {"series_id": sid, "start_date": start_date}

    return echo


if __name__ == "__main__":
    source = boc_config_demo()          # no arguments — injected from config.toml
    for row in source.resources["boc_settings_echo"]:
        print(row)

python day23/config_demo.py

dlt.config.value means "this argument must be supplied, either by the caller or by configuration". Explicit arguments always win: boc_config_demo(series_ids=["V39079"]) ignores the file for that argument.

section="boc" tells dlt to look in [sources.boc]. Without it, dlt uses the name of the Python module the source is defined in.

Example 2 — When a value is missing, read the error

Comment out start_date in config.toml and run again. The error message lists every place dlt looked: environment variables, secrets.toml, config.toml, from the most specific section to the most general. When configuration "doesn't work", this list tells you exactly what key name dlt expected.

Example 3 — Environment variables override files

Every setting can also come from an environment variable: take the TOML path, join the parts with double underscores, and use capitals:

SOURCES__BOC__START_DATE=2025-01-01 python day23/config_demo.py

The start date changed without touching any file.

Lookup priority (highest first): explicit function arguments → environment variables → .dlt/secrets.toml → .dlt/config.toml → default values in the function signature.

This order is what makes pipelines portable. Your committed config.toml holds sensible defaults, and each environment overrides just what it needs with environment variables. Schedulers such as Prefect (Days 51–60) and cloud platforms set environment variables, not files.

23.3 Secrets in Practice: the FRED API Key (23 min)

On Day 9 you may have registered a FRED API key. Now let dlt handle it.

Store it (choose one)

Option A — local file (quick, for your Codespace only). In .dlt/secrets.toml:

[sources.fred]
api_key = "paste-your-key-here"

Option B — Codespaces secret (recommended). On GitHub: Settings → Codespaces → Secrets → New secret, named SOURCES__FRED__API_KEY, with access to macro-data-course. Restart the Codespace; dlt now finds the key as an environment variable, with no file needed.

Use it

Create day23/fred_api_source.py:

from datetime import date

import dlt
from dlt.sources.helpers import requests

API = "https://api.stlouisfed.org/fred/series/observations"


@dlt.source(name="fred", section="fred")
def fred_api_source(series_ids: list = dlt.config.value,
                    api_key: str = dlt.secrets.value,
                    start_date: str = "2006-01-01"):

    @dlt.resource(name="fred_observations", write_disposition="replace",
                  primary_key=("series_id", "date"),
                  columns={"value": {"data_type": "double"}})
    def observations():
        for sid in series_ids:
            r = requests.get(API, params={
                "series_id": sid, "api_key": api_key, "file_type": "json",
                "observation_start": start_date}, timeout=30)
            r.raise_for_status()
            rows = [{"date": date.fromisoformat(o["date"]), "series_id": sid,
                     "value": float(o["value"]), "source": "fred"}
                    for o in r.json().get("observations", [])
                    if o.get("value") not in (None, "", ".")]
            yield rows

    return observations

Add the series list to config.toml:

[sources.fred]
series_ids = ["FEDFUNDS", "DCOILWTICO", "CANLOLITOAASTSAM"]

Run it with the practice pipeline and check that it loads.

Secrets hygiene checklist

• secrets.toml is in .gitignore, verified with git check-ignore
• Secrets are never printed, logged or put in URLs you save (Day 9)
• The key is typed as a secret (dlt.secrets.value), so dlt masks it in its own logs and traces
• If a secret is ever committed, rotate it immediately (generate a new key and revoke the old one). Deleting the file in a later commit is not enough, because Git history keeps it.

Students without a FRED key: keep using the key-free CSV resource from Day 21. The capstone below uses the CSV endpoint, so everyone can complete it.

23.4 Destinations from Configuration: Dev vs Prod (18 min)

So far destination="duckdb" wrote to <pipeline_name>.duckdb. The credentials of a DuckDB destination are simply the path to the database file, and they can come from configuration too.

In .dlt/secrets.toml (destination credentials are conventionally treated as secrets, even when they're just a path):

[destination.duckdb]
credentials = "macro.duckdb"

Now any pipeline with destination="duckdb" loads into your warehouse. To load somewhere else temporarily, for example a throwaway practice file, override it with an environment variable:

DESTINATION__DUCKDB__CREDENTIALS=dev_macro.duckdb python ingest.py    # "dev"
python ingest.py                                                      # "prod" = macro.duckdb

The same code runs in both environments; only configuration changes. On a real team, "prod" might be a cloud warehouse such as MotherDuck, BigQuery or Snowflake. With dlt, switching is a configuration change plus installing the destination's extra (e.g. pip install "dlt[motherduck]").

Dataset names keep layers apart

We load dlt output into its own schema, raw_dlt, so it can't collide with raw.* tables built by load_warehouse.py:

macro.duckdb
├── raw.*            ← Day 13 loader (Parquet from macro_extract)
├── raw_dlt.*        ← dlt pipelines (from today)
├── seeds.*, staging.*, intermediate.*, marts.*

On Days 28–30 the staging models will switch over to raw_dlt, and the old path can be retired.

23.5 Capstone: the macro_ingest Package (35 min)

Rebuild Day 10's macro_extract as a dlt package, fully driven by configuration.

Configuration: .dlt/config.toml

[runtime]
log_level = "WARNING"

[sources.boc]
series_ids = ["V39079", "FXCADUSD", "BD.CDN.2YR.DQ.YLD", "BD.CDN.10YR.DQ.YLD"]
start_date = "2006-01-01"

[sources.fred]
series_ids = ["FEDFUNDS", "DCOILWTICO", "CANLOLITOAASTSAM"]
start_date = "2006-01-01"

[sources.statcan]
vectors = [41690973, 2062815]
latest_n = 250

Package layout

mkdir -p macro_ingest
touch macro_ingest/__init__.py

Requirements:

1. macro_ingest/boc.py: boc_source(series_ids=dlt.config.value, start_date=dlt.config.value) with resources boc_observations and boc_series_metadata (Day 22.2), section="boc".
2. macro_ingest/fred.py: fred_source(...) with fred_observations (CSV endpoint, Day 21.7), section="fred".
3. macro_ingest/statcan.py: statcan_source(vectors=..., latest_n=...) with metadata and data resources (Day 22.6), section="statcan".
4. All observation tables share the columns date, series_id, value, source and the primary key (series_id, date).
5. ingest.py at the repository root:
   • creates the pipeline macro_ingest → duckdb → dataset raw_dlt;
   • runs each source separately, so one failing publisher doesn't stop the others (Day 9's design decision);
   • stores each run's load info in a table _load_info;
   • prints a summary and exits 0 (all ok), 2 (partial) or 1 (all failed), as on Day 10.

"""Bank of Canada Valet source."""
from datetime import date

import dlt
from dlt.sources.helpers import requests

BASE = "https://www.bankofcanada.ca/valet"


def _fetch(series_id, start_date):
    r = requests.get(f"{BASE}/observations/{series_id}/json",
                     params={"start_date": start_date}, timeout=30)
    r.raise_for_status()
    return r.json()


@dlt.source(name="boc", section="boc")
def boc_source(series_ids: list = dlt.config.value,
               start_date: str = dlt.config.value):

    @dlt.resource(name="boc_observations", write_disposition="replace",
                  primary_key=("series_id", "date"),
                  columns={"value": {"data_type": "double"}})
    def observations():
        for sid in series_ids:
            rows = []
            for obs in _fetch(sid, start_date).get("observations", []):
                cell = obs.get(sid)
                try:
                    value = float(cell.get("v"))
                except (AttributeError, TypeError, ValueError):
                    continue
                rows.append({"date": date.fromisoformat(obs["d"]), "series_id": sid,
                             "value": value, "source": "boc"})
            yield rows

    @dlt.resource(name="boc_series_metadata", write_disposition="replace",
                  primary_key="series_id")
    def series_metadata():
        for sid in series_ids:
            info = _fetch(sid, date.today().isoformat()).get("seriesDetail", {}).get(sid, {})
            yield {"series_id": sid, "label": info.get("label"),
                   "description": info.get("description")}

    return observations, series_metadata

"""FRED source (graph CSV endpoint, no key required)."""
import csv
import io
from datetime import date

import dlt
from dlt.sources.helpers import requests

URL = "https://fred.stlouisfed.org/graph/fredgraph.csv"


@dlt.source(name="fred", section="fred")
def fred_source(series_ids: list = dlt.config.value,
                start_date: str = dlt.config.value):

    @dlt.resource(name="fred_observations", write_disposition="replace",
                  primary_key=("series_id", "date"),
                  columns={"value": {"data_type": "double"}})
    def observations():
        start = date.fromisoformat(start_date)
        for sid in series_ids:
            r = requests.get(URL, params={"id": sid}, timeout=30)
            r.raise_for_status()
            reader = csv.reader(io.StringIO(r.text))
            next(reader)
            rows = []
            for line in reader:
                if len(line) < 2 or line[1] in ("", "."):
                    continue
                d = date.fromisoformat(line[0])
                if d >= start:
                    rows.append({"date": d, "series_id": sid,
                                 "value": float(line[1]), "source": "fred"})
            yield rows

    return observations

"""Statistics Canada WDS vector source."""
from datetime import date

import dlt
from dlt.sources.helpers import requests

WDS = "https://www150.statcan.gc.ca/t1/wds/rest"


@dlt.source(name="statcan", section="statcan")
def statcan_source(vectors: list = dlt.config.value,
                   latest_n: int = dlt.config.value):

    @dlt.resource(name="statcan_series_metadata", write_disposition="replace",
                  primary_key="vector_id")
    def series_metadata():
        r = requests.post(f"{WDS}/getSeriesInfoFromVector",
                          json=[{"vectorId": v} for v in vectors], timeout=30)
        r.raise_for_status()
        for item in r.json():
            if item.get("status") == "SUCCESS":
                obj = item["object"]
                yield {"vector_id": f"v{obj['vectorId']}",
                       "product_id": obj.get("productId"),
                       "title": obj.get("SeriesTitleEn"),
                       "frequency_code": obj.get("frequencyCode")}

    @dlt.resource(name="statcan_observations", write_disposition="replace",
                  primary_key=("series_id", "date"),
                  columns={"value": {"data_type": "double"}})
    def observations():
        body = [{"vectorId": v, "latestN": latest_n} for v in vectors]
        r = requests.post(f"{WDS}/getDataFromVectorsAndLatestNPeriods",
                          json=body, timeout=60)
        r.raise_for_status()
        for item in r.json():
            if item.get("status") != "SUCCESS":
                continue
            obj = item["object"]
            sid = f"v{obj['vectorId']}"
            yield [{"date": date.fromisoformat(dp["refPer"][:10]), "series_id": sid,
                    "value": float(dp["value"]), "source": "statcan"}
                   for dp in obj.get("vectorDataPoint", [])
                   if dp.get("value") is not None]

    return series_metadata, observations

"""ingest.py — load all macro sources into DuckDB with dlt.

Destination file comes from .dlt/secrets.toml ([destination.duckdb] credentials)
or the DESTINATION__DUCKDB__CREDENTIALS environment variable.

Usage:  python ingest.py
"""
import sys

import dlt

from macro_ingest.boc import boc_source
from macro_ingest.fred import fred_source
from macro_ingest.statcan import statcan_source

SOURCES = {"boc": boc_source, "fred": fred_source, "statcan": statcan_source}


def main():
    pipeline = dlt.pipeline(pipeline_name="macro_ingest",
                            destination="duckdb",
                            dataset_name="raw_dlt")
    ok, failed = [], []

    for name, make_source in SOURCES.items():
        try:
            info = pipeline.run(make_source())       # config injected here
            info.raise_on_failed_jobs()
            pipeline.run([info], table_name="_load_info")
            ok.append(name)
            print(f"✓ {name}")
        except Exception as e:
            failed.append(name)
            print(f"✗ {name}: {e}")

    print(f"\n{len(ok)} source(s) loaded, {len(failed)} failed {failed}")
    if not ok:
        sys.exit(1)
    if failed:
        sys.exit(2)


if __name__ == "__main__":
    main()

Run it and compare with the old pipeline

python ingest.py
duckdb macro.duckdb -readonly

.tables

-- Parity check: dlt vs the Day 10 pipeline, per series
WITH new AS (
    SELECT series_id, COUNT(*) AS n_dlt, MAX(date) AS latest_dlt
    FROM (SELECT * FROM raw_dlt.boc_observations
          UNION ALL BY NAME SELECT * FROM raw_dlt.fred_observations
          UNION ALL BY NAME SELECT * FROM raw_dlt.statcan_observations)
    GROUP BY series_id
),
old AS (
    SELECT series_id, COUNT(*) AS n_old, CAST(MAX(date) AS DATE) AS latest_old
    FROM raw.observations
    GROUP BY series_id
)
SELECT COALESCE(new.series_id, old.series_id) AS series_id,
       n_old, n_dlt, latest_old, latest_dlt
FROM new FULL OUTER JOIN old USING (series_id)
ORDER BY series_id;

Small differences are expected: data published between the two runs, or different start dates for StatCan vectors. Large differences mean a bug in one of the two paths. Investigate before moving on.

23.6 Smoke Tests · Recap (10 min)

A smoke test is a quick run that checks the machinery works end to end without doing all the work. Combine three things you learned: add_limit, a throwaway destination, and a small config override.

Create day23/smoke_test.py:

"""Fast end-to-end check: tiny loads into a throwaway DuckDB file."""
import os

import dlt

from macro_ingest.boc import boc_source
from macro_ingest.fred import fred_source
from macro_ingest.statcan import statcan_source

os.environ["DESTINATION__DUCKDB__CREDENTIALS"] = "smoke_test.duckdb"

pipeline = dlt.pipeline(pipeline_name="macro_smoke", destination="duckdb",
                        dataset_name="smoke", dev_mode=True)

for source in (boc_source(start_date="2026-01-01"),
               fred_source(start_date="2026-01-01"),
               statcan_source(latest_n=3)):
    info = pipeline.run(source.add_limit(1))
    info.raise_on_failed_jobs()
    print(f"✓ {source.name}")

print("Smoke test passed.")

python day23/smoke_test.py && rm -f smoke_test.duckdb

• Explicit arguments (start_date=...) override config.toml for this run only.
• dev_mode=True gives every run a fresh, uniquely named dataset, which is ideal for experiments that shouldn't interfere with each other.

Run this before ingest.py whenever you change the package: it takes seconds instead of minutes.

Recap

Looking ahead: Days 24–27

• Schema evolution: what happens when an API adds, renames or changes a field, and how to control it with schema contracts.
• Flattening nested JSON: loading raw Valet and WDS responses as-is and letting dlt create parent/child tables.
• Incremental loading: fetching only new observations with dlt.sources.incremental, and using merge to absorb StatCan revisions.

Day 23 checklist

• I know what belongs in config.toml vs secrets.toml, and my secrets file is ignored by Git
• I can inject settings with dlt.config.value and dlt.secrets.value, and read dlt's lookup errors
• I can override any setting with a correctly named environment variable
• I can point the same pipeline at different DuckDB files (dev vs prod) through configuration
• My macro_ingest package loads BoC, FRED and StatCan into macro.duckdb with one command
• I verified parity with the Day 10 pipeline and have a fast smoke test

git add .
git status                         # .dlt/secrets.toml must NOT appear
git commit -m "Day 23: dlt config & secrets, macro_ingest package"
git push