dlt's REST Toolkit

Reuse the practice helpers from Day 24 (common.py with practice_pipeline and reset). Copy it into day28/, day29/ and day30/.

28.1 RESTClient (25 min)

So far every resource built URLs and called requests.get itself. RESTClient bundles the things every call to one API shares (base URL, headers, authentication, pagination) into a single configured object. It uses dlt's retrying session underneath.

Create day28/rest_client_basics.py:

from dlt.sources.helpers.rest_client import RESTClient
from dlt.sources.helpers.rest_client.paginators import SinglePagePaginator

valet = RESTClient(
    base_url="https://www.bankofcanada.ca/valet/",
    headers={"User-Agent": "macro-data-course/1.0 (educational use)"},
    paginator=SinglePagePaginator(),        # Valet returns everything in one response
)

# 1. A single request
resp = valet.get("observations/V39079/json", params={"recent": 3})
resp.raise_for_status()
print(resp.json()["observations"])

# 2. paginate(): yields pages of *items*, extracted with a data selector
for page in valet.paginate("observations/FXCADUSD/json",
                           params={"recent": 5},
                           data_selector="observations"):
    print(type(page), len(page))            # one page (a list) of observation dicts
    print(page[0])

JSONPath in one table

Why bother?

• One place to configure each API: change a header or timeout once, not in every resource.
• Retries come from dlt's session (configured on Day 29).
• Pagination becomes a parameter instead of a hand-written loop.
• The same client works in code-based resources and the declarative source (28.4).

28.2 Authentication Helpers (20 min)

dlt ships authentication classes so secrets are attached consistently, and never appear in your resource code.

import dlt
from dlt.sources.helpers.rest_client import RESTClient
from dlt.sources.helpers.rest_client.auth import APIKeyAuth

fred = RESTClient(
    base_url="https://api.stlouisfed.org/fred/",
    auth=APIKeyAuth(name="api_key",
                    api_key=dlt.secrets["sources.fred.api_key"],   # Day 23
                    location="query"),                             # ?api_key=... on every call
)

resp = fred.get("series/observations",
                params={"series_id": "FEDFUNDS", "file_type": "json", "limit": 3,
                        "sort_order": "desc"})
resp.raise_for_status()
print(resp.json()["observations"])

Students without a FRED key: read along. Every exercise today also has a keyless alternative (Valet, World Bank, StatCan).

28.3 Pagination (30 min)

Many APIs return large results in pages: records 1–100, then 101–200, and so on. The API tells you how to get the next page, and a paginator knows how to follow those instructions.

A keyless, paginated, economic API: the World Bank

The World Bank Indicators API returns a two-element list: metadata (including the total number of pages), then the records:

curl -s "https://api.worldbank.org/v2/country/CAN/indicator/NY.GDP.MKTP.KD.ZG?format=json&per_page=3&page=1" | python -m json.tool | head -40

[
  {"page": 1, "pages": 22, "per_page": 3, "total": 65, ...},   ← $[0]
  [ {"country": {...}, "date": "2024", "value": 1.5, ...}, ... ] ← $[1]
]

Create day28/world_bank.py:

from dlt.sources.helpers.rest_client import RESTClient
from dlt.sources.helpers.rest_client.paginators import PageNumberPaginator

wb = RESTClient(
    base_url="https://api.worldbank.org/v2/",
    paginator=PageNumberPaginator(
        base_page=1,              # the first page is 1, not 0
        page_param="page",        # the query parameter holding the page number
        total_path="$[0].pages",  # where the total number of pages is found
    ),
)

G7 = "CAN;USA;GBR;DEU;FRA;ITA;JPN"
n_pages = 0
records = []
for page in wb.paginate(f"country/{G7}/indicator/NY.GDP.MKTP.KD.ZG",
                        params={"format": "json", "per_page": 50, "date": "2000:2025"},
                        data_selector="$[1]"):
    n_pages += 1
    records.extend(page)

print(f"{n_pages} pages, {len(records)} records")
print(records[0])

per_page=50 is deliberately small so you see several pages. In production, use the largest page size the API allows, because fewer requests means faster runs and fewer chances to fail.

Try it 28.3

Turn the loop above into a @dlt.resource named wb_gdp_growth that yields each page, with write_disposition="replace". Load it into a practice pipeline, then query the average real GDP growth by country for 2015–2019 vs 2021–2024. (Remember Day 26: the nested country object becomes country__id / country__value columns.)

import dlt
from common import practice_pipeline, reset


@dlt.resource(name="wb_gdp_growth", write_disposition="replace")
def wb_gdp_growth():
    yield from wb.paginate(f"country/{G7}/indicator/NY.GDP.MKTP.KD.ZG",
                           params={"format": "json", "per_page": 100, "date": "2000:2025"},
                           data_selector="$[1]")


reset("rest_lab")
p = practice_pipeline("rest_lab", "lab")
print(p.run(wb_gdp_growth()))

SELECT country__value AS country,
       ROUND(AVG(value) FILTER (WHERE CAST(date AS INT) BETWEEN 2015 AND 2019), 2) AS avg_2015_19,
       ROUND(AVG(value) FILTER (WHERE CAST(date AS INT) BETWEEN 2021 AND 2024), 2) AS avg_2021_24
FROM lab.wb_gdp_growth
GROUP BY country
ORDER BY avg_2021_24 DESC;

28.4 The Declarative REST API Source (23 min)

For many APIs you don't need to write resource code at all. You describe the API in a dictionary, and dlt's rest_api_source builds the resources.

Example 1 — World Bank, declaratively

from dlt.sources.rest_api import rest_api_source
from common import practice_pipeline

config = {
    "client": {
        "base_url": "https://api.worldbank.org/v2/",
        "paginator": {"type": "page_number", "base_page": 1,
                      "page_param": "page", "total_path": "$[0].pages"},
    },
    "resource_defaults": {"write_disposition": "replace"},
    "resources": [
        {
            "name": "wb_gdp_growth_decl",
            "endpoint": {
                "path": "country/CAN;USA;GBR;DEU;FRA;ITA;JPN/indicator/NY.GDP.MKTP.KD.ZG",
                "params": {"format": "json", "per_page": 100, "date": "2000:2025"},
                "data_selector": "$[1]",
            },
        },
    ],
}

p = practice_pipeline("rest_lab", "lab")
print(p.run(rest_api_source(config)))

Same result as 28.3, with zero resource code. The configuration is the extractor.

Example 2 — Valet with a declarative incremental cursor

The REST API source can also manage an incremental cursor (Day 27): it fills a query parameter with the last value seen.

config = {
    "client": {"base_url": "https://www.bankofcanada.ca/valet/", "paginator": "single_page"},
    "resources": [
        {
            "name": "valet_overnight_raw",
            "write_disposition": "merge",
            "primary_key": "d",
            "endpoint": {
                "path": "observations/V39079/json",
                "data_selector": "observations",
                "params": {
                    "start_date": {"type": "incremental", "cursor_path": "d",
                                   "initial_value": "2025-01-01"},
                },
            },
        },
    ],
}
p.run(rest_api_source(config))

Run it twice. The second run requests only data after the last d it loaded. The table keeps Valet's raw shape (d, and a flattened column for V39079.v), which a staging view would clean (Day 26.4).

Declarative or code?

For our core BoC and StatCan extractors, we'll keep code-based resources but build them on RESTClient. For new, well-behaved APIs, start declarative.

The REST API source also supports dependent resources (the declarative version of Day 22's transformers), processing steps and many paginator options. Its documentation is worth browsing once you're comfortable with the basics.

28.5 POST Endpoints: StatCan WDS Through RESTClient (15 min)

StatCan's vector endpoints use POST with a JSON body (Day 10). RESTClient handles that too:

from dlt.sources.helpers.rest_client import RESTClient
from dlt.sources.helpers.rest_client.paginators import SinglePagePaginator

wds = RESTClient(base_url="https://www150.statcan.gc.ca/t1/wds/rest/",
                 paginator=SinglePagePaginator())

resp = wds.post("getDataFromVectorsAndLatestNPeriods",
                json=[{"vectorId": 41690973, "latestN": 3},
                      {"vectorId": 999999999, "latestN": 3}])     # a vector that doesn't exist
resp.raise_for_status()
for item in resp.json():
    print(item["status"], str(item["object"])[:80])

Notice: HTTP 200, yet one item says FAILED. raise_for_status() can't catch this, so your code must check each item's status. This is the most important StatCan-specific resilience rule, and we build it in on Day 29.

28.6 Mini-Project: A Declarative G7 GDP-Growth Source (23 min)

Extend the World Bank configuration to three indicators, each as its own resource, loaded into one dataset:

Requirements:

1. Build the resources list with a Python loop over a dictionary of {name: indicator_code}, so adding an indicator is a one-line change.
2. Put the country list and date range in .dlt/config.toml under [sources.world_bank] and read them with dlt.config["sources.world_bank.countries"], etc.
3. After loading, write one SQL query producing a long table (country, year, indicator, value) across all three resources.
4. Which G7 economy had the highest average inflation in 2021–2023?

# .dlt/config.toml
[sources.world_bank]
countries = "CAN;USA;GBR;DEU;FRA;ITA;JPN"
date_range = "2000:2025"

import dlt
from dlt.sources.rest_api import rest_api_source
from common import practice_pipeline

INDICATORS = {
    "wb_gdp_growth": "NY.GDP.MKTP.KD.ZG",
    "wb_inflation": "FP.CPI.TOTL.ZG",
    "wb_unemployment": "SL.UEM.TOTL.ZS",
}
countries = dlt.config["sources.world_bank.countries"]
date_range = dlt.config["sources.world_bank.date_range"]

config = {
    "client": {
        "base_url": "https://api.worldbank.org/v2/",
        "paginator": {"type": "page_number", "base_page": 1,
                      "page_param": "page", "total_path": "$[0].pages"},
    },
    "resource_defaults": {"write_disposition": "replace"},
    "resources": [
        {"name": name,
         "endpoint": {"path": f"country/{countries}/indicator/{code}",
                      "params": {"format": "json", "per_page": 1000, "date": date_range},
                      "data_selector": "$[1]"}}
        for name, code in INDICATORS.items()
    ],
}

p = practice_pipeline("rest_lab", "lab")
print(p.run(rest_api_source(config, name="world_bank")))

WITH long AS (
    SELECT country__value AS country, CAST(date AS INT) AS year,
           'gdp_growth' AS indicator, value FROM lab.wb_gdp_growth
    UNION ALL
    SELECT country__value, CAST(date AS INT), 'inflation', value FROM lab.wb_inflation
    UNION ALL
    SELECT country__value, CAST(date AS INT), 'unemployment', value FROM lab.wb_unemployment
)
SELECT country, ROUND(AVG(value), 2) AS avg_inflation_2021_23
FROM long
WHERE indicator = 'inflation' AND year BETWEEN 2021 AND 2023
GROUP BY country
ORDER BY avg_inflation_2021_23 DESC;

Day 28 checklist

• I can configure a RESTClient with a base URL, headers, a paginator and a data selector
• I attach API keys with dlt's auth helpers instead of hand-building URLs
• I can recognise pagination styles and configure the matching paginator
• I can describe an API declaratively with rest_api_source, including an incremental cursor
• I can call POST endpoints and I know that HTTP 200 doesn't mean every item succeeded
• I can decide when declarative beats code, and vice versa

git add .
git commit -m "Day 28: RESTClient, auth, pagination, declarative REST sources"
git push