Sources vs Resources

22.1 The Design Idea (20 min)

A source is like a publisher, and its resources are the publications. You configure the publisher once (API key, base URL, start date) and every publication uses it.

For our project:

source: boc           → resources: boc_observations, boc_series_metadata
source: fred          → resources: fred_observations
source: statcan       → resources: statcan_vector_data, statcan_series_metadata

Keeping metadata as a separate resource (labels, units, frequency) gives you the dimension table from the Day 12 star schema for free.

22.2 @dlt.source: Grouping BoC Resources (25 min)

Create day22/boc_source.py:

from datetime import date

import dlt
from dlt.sources.helpers import requests

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


def _fetch(series_id, start_date):
    """Download one series; return (payload) — shared by both resources."""
    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")
def boc_source(series_ids, start_date="2006-01-01"):
    """Bank of Canada Valet API: observations + series metadata."""

    @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:
            payload = _fetch(sid, start_date)
            rows = []
            for obs in payload.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:
            payload = _fetch(sid, date.today().isoformat())   # tiny request: today only
            info = payload.get("seriesDetail", {}).get(sid, {})
            yield {"series_id": sid,
                   "label": info.get("label"),
                   "description": info.get("description")}

    return observations, series_metadata


if __name__ == "__main__":
    pipeline = dlt.pipeline(pipeline_name="dlt_practice",
                            destination="duckdb", dataset_name="practice")
    source = boc_source(["V39079", "FXCADUSD", "BD.CDN.2YR.DQ.YLD", "BD.CDN.10YR.DQ.YLD"])
    print(pipeline.run(source))

Things to notice:

• The resources are defined inside the source function, so they can use series_ids and start_date directly. Python calls this a closure.
• _fetch starts with an underscore, a Python convention meaning "internal helper, not for outside use".
• The source returns its resources, and pipeline.run(source) loads them all.

SELECT m.label, o.series_id, COUNT(*) AS n, MAX(o.date) AS latest
FROM practice.boc_observations AS o
JOIN practice.boc_series_metadata AS m USING (series_id)
GROUP BY ALL;

A fact table and a dimension table from one source, joined in SQL: Day 12's star schema, now fed automatically.

22.3 Table Design: One Long Table or One Table per Series? (23 min)

You could also create one table per series (boc_v39079, boc_fxcadusd, …) by creating resources dynamically:

@dlt.source(name="boc_wide")
def boc_per_series(series_ids, start_date="2006-01-01"):
    resources = []
    for sid in series_ids:
        def make(sid=sid):                     # sid=sid "freezes" the loop value
            payload = _fetch(sid, start_date)
            for obs in payload.get("observations", []):
                cell = obs.get(sid)
                try:
                    yield {"date": date.fromisoformat(obs["d"]),
                           "value": float(cell.get("v"))}
                except (AttributeError, TypeError, ValueError):
                    continue
        resources.append(dlt.resource(make(), name=f"boc_{sid}",
                                      write_disposition="replace"))
    return resources

The sid=sid trick: without it, every inner function would see the last value of sid when it finally runs, because generators are lazy (Day 21). Binding it as a default argument captures the value at creation time. This is a classic Python gotcha.

Which design is better?

For homogeneous time series, long is almost always better, and it's what the rest of the course uses. Per-resource tables make sense when records have genuinely different structures (e.g. BoC decisions vs yield observations).

22.4 Selecting, Limiting and Transforming Resources (23 min)

Selecting some resources from a source

source = boc_source(["V39079"])
print(list(source.resources.keys()))       # ['boc_observations', 'boc_series_metadata']

pipeline.run(source.with_resources("boc_series_metadata"))   # load only metadata

Limiting during development

When experimenting, don't download 20 years of data each time:

pipeline.run(boc_source(["V39079", "FXCADUSD"]).add_limit(1))

add_limit(1) stops each resource after its first yield. With our code that means one series' page, which is ideal for quick tests.

Transforming items with add_map and add_filter

These apply a function to every item as it flows through, without editing the resource itself:

def add_units(row):
    row["unit"] = "percent" if row["series_id"] == "V39079" else "CAD per USD"
    return row

def recent_only(row):
    return row["date"] >= date(2020, 1, 1)

source = boc_source(["V39079", "FXCADUSD"])
source.resources["boc_observations"].add_map(add_units).add_filter(recent_only)
pipeline.run(source)

Where should logic live? Light, source-specific cleaning (units, dropping unusable rows) can live in dlt. Economic transformations (growth rates, aggregation) belong in the SQL layers from Days 18–20, and later in dbt. Keep ingestion boring.

Try it 22.4

Add a map step that converts BoC percent values into basis points for V39079 only, storing the result in a new column value_bp. Load with add_limit(1) and check the new column in DuckDB.

def add_bp(row):
    if row["series_id"] == "V39079":
        row["value_bp"] = row["value"] * 100
    return row

source = boc_source(["V39079"])
source.resources["boc_observations"].add_map(add_bp)
pipeline.run(source.add_limit(1))

SELECT date, value, value_bp FROM practice.boc_observations ORDER BY date DESC LIMIT 3;

22.5 Transformers: Dependent Resources (23 min)

Sometimes one resource's output drives another's requests: first list the series in a group, then fetch each series. A transformer receives items from a parent resource.

Create day22/boc_groups.py:

from datetime import date

import dlt
from dlt.sources.helpers import requests

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


@dlt.resource(name="boc_group_series", write_disposition="replace",
              primary_key="series_id")
def boc_group_series(groups):
    """Yield one metadata record per series in each BoC group."""
    for g in groups:
        r = requests.get(f"{BASE}/observations/group/{g}/json",
                         params={"recent": 1}, timeout=30)
        r.raise_for_status()
        for sid, info in r.json().get("seriesDetail", {}).items():
            yield {"series_id": sid, "group_name": g,
                   "label": info.get("label"), "description": info.get("description")}


@dlt.transformer(name="boc_group_observations", write_disposition="replace",
                 primary_key=("series_id", "date"),
                 columns={"value": {"data_type": "double"}})
def boc_group_observations(series, start_date="2024-01-01"):
    """For each series record received, fetch its observations."""
    sid = series["series_id"]
    r = requests.get(f"{BASE}/observations/{sid}/json",
                     params={"start_date": start_date}, timeout=30)
    r.raise_for_status()
    rows = []
    for obs in r.json().get("observations", []):
        cell = obs.get(sid)
        try:
            rows.append({"date": date.fromisoformat(obs["d"]), "series_id": sid,
                         "value": float(cell.get("v"))})
        except (AttributeError, TypeError, ValueError):
            continue
    yield rows


@dlt.source(name="boc_groups")
def boc_groups_source(groups, start_date="2024-01-01"):
    catalogue = boc_group_series(groups)
    return catalogue, catalogue | boc_group_observations(start_date=start_date)


if __name__ == "__main__":
    pipeline = dlt.pipeline(pipeline_name="dlt_practice",
                            destination="duckdb", dataset_name="practice")
    print(pipeline.run(boc_groups_source(["BOND_YIELDS_BENCHMARK"])))

The pipe operator | connects a resource to a transformer: every item the catalogue yields flows into boc_group_observations. Because the source returns both the catalogue and the piped transformer, both tables are loaded, and the catalogue is fetched only once.

SELECT s.label, COUNT(*) AS n, ROUND(arg_max(o.value, o.date), 2) AS latest
FROM practice.boc_group_observations AS o
JOIN practice.boc_group_series AS s USING (series_id)
GROUP BY s.label
ORDER BY s.label;

Transformers shine when APIs are hierarchical: tables → series → observations (StatCan), groups → series → observations (BoC), or pages of search results → detail pages.

22.6 Mini-Project: A Statistics Canada Source (23 min)

Create day22/statcan_source.py with a source statcan_source(vectors, latest_n=250) containing two resources:

1. statcan_series_metadata: POSTs to getSeriesInfoFromVector (Day 10) and yields {vector_id, product_id, title, frequency_code} per vector.
2. statcan_vector_data: POSTs to getDataFromVectorsAndLatestNPeriods and yields long rows {date, series_id, value, source}, where series_id is like v41690973.

Load vectors 41690973 (CPI All-items) and 2062815 (unemployment rate, as verified on Day 10), then join data to metadata in SQL.

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")
def statcan_source(vectors, latest_n=250):

    @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":
                continue
            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_vector_data", write_disposition="replace",
                  primary_key=("series_id", "date"),
                  columns={"value": {"data_type": "double"}})
    def vector_data():
        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']}"
            rows = []
            for dp in obj.get("vectorDataPoint", []):
                if dp.get("value") is None:
                    continue
                rows.append({"date": date.fromisoformat(dp["refPer"][:10]),
                             "series_id": sid, "value": float(dp["value"]),
                             "source": "statcan"})
            yield rows

    return series_metadata, vector_data


if __name__ == "__main__":
    pipeline = dlt.pipeline(pipeline_name="dlt_practice",
                            destination="duckdb", dataset_name="practice")
    print(pipeline.run(statcan_source([41690973, 2062815])))

SELECT m.title, d.series_id, COUNT(*) AS n, MAX(d.date) AS latest
FROM practice.statcan_vector_data AS d
JOIN practice.statcan_series_metadata AS m ON m.vector_id = d.series_id
GROUP BY ALL;

Day 22 checklist

• I can explain the difference between a source and a resource, with examples
• I can write a @dlt.source that returns several resources sharing arguments
• I can argue for long vs per-series table designs, and avoid the loop-variable gotcha
• I can select resources, limit them for development, and use add_map/add_filter
• I can chain a resource into a transformer with |
• I load metadata alongside data to get dimension tables automatically

git add .
git commit -m "Day 22: dlt sources, transformers, StatCan source"
git push