Fetch Contract

Your fetch() function is called once per page by the runtime. It receives pagination context and returns rows. Starlark handles I/O — SQL handles transformation.

The @fetch directive

A model that calls a lib function in its FROM clause must declare @fetch:

-- @kind: append
-- @fetch
-- @incremental: date

SELECT
    series::VARCHAR AS series,
    date::DATE AS date,
    value::DECIMAL(18,6) AS value
FROM riksbank('SEKEURPMI')

@fetch is a bare marker (no arguments). Arguments to the lib call go on the FROM line, not the directive. The directive activates the strict-fetch validator and signals “this model produces its schema from a lib call”.

A @fetch model and a @push model are two different things — they cannot be combined on the same model. To push data fetched from one API to another, split into a @fetch model that materializes the raw rows and a downstream @push model that reads from the materialized table.

Function signature

def fetch(arg1, arg2, ..., page, columns=[], target="", is_backfill=True):

Arguments before page come from API.fetch.args. They receive values from the SQL call:

API = {"fetch": {"args": ["resource"]}}

SELECT id::BIGINT AS id, name::VARCHAR AS name FROM my_api('users')
--                                                          ↑
--                                                       resource

Declare only the parameters your blueprint uses. The runtime automatically filters kwargs to match your function signature — undeclared kwargs are silently dropped, not errors.

Runtime-injected kwargs

All are optional — declare them with defaults if your blueprint needs them. Incremental kwargs (is_backfill, last_value, etc.) are only meaningful when the SQL model uses @incremental.

Typed columns

columns is how SQL communicates schema intent to Starlark. Each entry is a dict with name and type. The type is in DuckDB-native syntax — the same shape DuckDB’s json_structure produces and json_transform consumes.

columns = [
    {"name": "invoice_number", "type": "VARCHAR"},
    {"name": "total",          "type": "DECIMAL(18,3)"},
    {"name": "date",           "type": "DATE"},
    {"name": "created_at",     "type": "TIMESTAMPTZ"},
    {"name": "items",          "type": "JSON"},
    {"name": "tags",           "type": ["VARCHAR"]},
    {"name": "address",        "type": {"street": "VARCHAR", "zip": "INTEGER"}},
]

Width and precision distinctions are preserved (TINYINT vs BIGINT, FLOAT vs DOUBLE, TIMESTAMP_NS vs TIMESTAMPTZ) — the column dict carries enough information to round-trip back to the same DuckDB type.

The name field is the cast source column reference, not the SQL alias. For

SELECT AD_UNIT_NAME::VARCHAR AS ad_unit FROM gam_report()

the blueprint sees name = "AD_UNIT_NAME" (the API field it queries by and uses as the row key). The alias ad_unit is the materialized column name in DuckLake; the runtime renames at projection time without involving the blueprint. This lets a model author rename API fields without modifying the lib.

JSON Schema mapping

Blueprints that talk to JSON-Schema-aware APIs (OpenAI structured outputs, Mistral OCR, Anthropic) use the lib_helpers.to_json_schema helper to convert a DuckDB type into a JSON Schema dict:

def fetch(page, columns=[]):
    schema = {"type": "object", "properties": {
        col["name"]: lib_helpers.to_json_schema(col["type"])
        for col in columns
    }}
    resp = http.post("/v1/structured", json={"schema": schema, ...})

Width/precision distinctions inside DuckDB collapse to the same JSON Schema type — JSON Schema doesn’t carry that information.

Blueprints that don’t need a JSON Schema (Riksbank, Google Sheets fetch) ignore the type field entirely.

Page object

Read-only struct:

SQL schema contract

A @fetch model is a passthrough projection of API rows. The strict-fetch validator runs before any fetch() call and rejects any model that doesn’t fit the canonical shape:

-- @fetch
SELECT
    col1::TYPE AS alias1,
    col2::TYPE AS alias2,
    ...
FROM lib(args)

That’s it. Every constraint below is mechanical:

Validation runs at the start of every ondatrasql run / ondatrasql sandbox. A violation aborts the run before any fetch() is called, naming the offending projection.

Two consequences worth knowing:

• Empty fetches produce a typed target from the SQL alone — the runtime never has to guess types on the first run or after a schema change.
• Schema evolution is detected from the SQL diff, not from runtime data inspection. Add a column to the SELECT and the next run alters the target.

The fix for almost every rejection is the two-model pattern below.

Return format

return {
    "rows": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}],
    "next": "cursor_abc123",
}

The cursor is opaque — the runtime stores and returns it as-is. Dicts pass through directly:

next_cursor = {"idx": series_idx, "from": next_date, "until": yesterday}

Empty fetches and tracked / scd2

For tracked- and scd2-kind models the runtime needs to know what a 0-row fetch means. Two cases:

1. Source has nothing new this run — the lib reads a cache key, last-modified timestamp, or pre-computed digest, decides nothing changed upstream, and returns 0 rows. The target should stay as-is.
2. Source is fully enumerated and is genuinely empty — every row in the target should be deleted (tracked: missing groups; scd2: current versions are closed).

Set empty_result on the final page (the one whose next is None / missing / "") to disambiguate. The default is no_change, which is safe for fetches that may return 0 rows for any reason. Use delete_missing only when the lib enumerates the full source on every run and an empty fetch is the authoritative “everything is gone” signal.

def fetch(page):
    if cache_key_unchanged():
        return {"rows": [], "next": None}  # implicit empty_result=no_change

def fetch(page):
    rows = http.get("/items").json
    return {"rows": rows, "next": None, "empty_result": "delete_missing"}

empty_result has no effect when the fetch returned any rows in total. Unknown string values fall back to the default rather than failing the run.

Pagination

The runtime calls fetch() in a loop until "next" is None or missing.

Pagination patterns vary by API. The blueprint owns the logic:

• Cursor-based — API returns a next token, pass it back
• Offset-based — increment an offset by page size
• Date-range — advance a date window per page
• Multi-resource — iterate through resources, then dates within each

The runtime doesn’t care which pattern you use — it just follows the cursor.

Async fetch

For APIs that create reports asynchronously (submit → poll → fetch results), declare async: True in the API dict and implement three functions instead of fetch():

API = {
    "fetch": {
        "async": True,
        "poll_interval": "5s",
        "poll_timeout": "5m",
        "poll_backoff": 2,
    },
}

def submit(columns=[], is_backfill=True, last_value="", initial_value=""):
    # Create and start the report/job
    return {"job_id": resp.json["id"]}  # opaque job reference

def check(job_ref):
    # Poll job status. Return None to keep polling, dict when done.
    if not done:
        return None
    return {"download_url": resp.json["result_url"]}  # opaque result reference

def fetch_result(result_ref, page):
    # Fetch result rows with pagination — same return format as fetch()
    return {"rows": [...], "next": next_token}

The runtime handles the poll loop — check() is called with the configured interval and backoff until it returns a non-None value or the timeout expires.

abort() in submit() is valid — materializes with 0 rows. fail() in any function stops the pipeline.

Error handling

Always check resp.ok. http.* only raises on 5xx, 429, and transport failures (those retry, then fail the run). A 4xx returns a response with resp.ok == False — it does not raise. Read resp.json without checking and a client error becomes a silent 0-row return, which for a table or scd2 model wipes / closes the target. ondatrasql validate reports an INFO finding (blueprint.unchecked_status) for any fetch that calls http.* without checking resp.ok / resp.status_code, and the runtime warns at run time; silence an intentional case (e.g. an API that 404s for an empty collection) with a # ondatracheck:allow-unchecked-status <reason> comment in the lib file.

Fail — stops the pipeline with an error:

if not resp.ok:
    fail("API error: " + str(resp.status_code) + " " + resp.text)

Abort — clean exit, 0 rows, no error. Use when there’s nothing to fetch:

if start_date > yesterday:
    abort()

The two-model pattern

@fetch models stay close to the API shape. Transformation — JOIN, WHERE, GROUP BY, derived expressions, multi-source combination — happens in downstream models that read from the materialized fetch table.

Raw model (@fetch) — column refs match the API:

-- models/raw/data.sql
-- @kind: append
-- @fetch
-- @incremental: date

SELECT
    series::VARCHAR AS series,
    date::DATE AS date,
    value::DECIMAL(18,6) AS value
FROM my_api('SERIES_A,SERIES_B')

Staging model — SQL transforms, casts types, pivots, joins:

-- models/staging/data.sql
-- @kind: table

SELECT
    date::DATE AS date,
    MAX(CASE WHEN series = 'SERIES_A' THEN value END)::DECIMAL AS series_a,
    MAX(CASE WHEN series = 'SERIES_B' THEN value END)::DECIMAL AS series_b
FROM raw.data
GROUP BY date

Joining two libs — write two raw models, then join in staging:

-- models/raw/users.sql
-- @kind: table
-- @fetch
SELECT user_id::BIGINT AS user_id, name::VARCHAR AS name FROM users_api('users')

-- models/raw/scores.sql
-- @kind: table
-- @fetch
SELECT user_id::BIGINT AS user_id, score::BIGINT AS score FROM scores_api('scores')

-- models/staging/joined.sql
-- @kind: table
SELECT
    u.user_id::BIGINT AS user_id,
    u.name::VARCHAR AS name,
    s.score::BIGINT AS score
FROM raw.users u
JOIN raw.scores s ON u.user_id = s.user_id

The split is mechanical and worth it: each @fetch model has its own lib lifecycle, its own incremental state, and a deterministic shape contract. The downstream model is plain SQL.

Available builtins and modules

Modules

DuckDB-backed builtins

Go-native builtins

Control flow