Python sources

Python sources turn any API, SaaS product or custom system into tables:

• Any source as a table. A Python function that fetches the data becomes a table, declared with sl.source and @source.table.
• The same loader as every other source. Python tables take part in plan, apply and load, with the same write strategies, cursors and contracts as Postgres tables and files.

They need the sqlload package (Install); sqlload-cli doesn’t run them.

REST APIs run in Rust

For a predictable REST API (paginated, fanned out over keys, retried, rows picked out of the JSON) describe the job with sl.http. Python only describes it; SQLLoad runs it entirely in Rust: requests, pages, retries, JSON parsing and conversion to Arrow. No row crosses between Python and Rust, so there is no per-row Python cost.

import sqlload as sl

shop = sl.source("shop", connection="shop", write_strategy="merge")
api = sl.http(shop, auth=sl.bearer(), rate_limit="10/s", retries=5)


@shop.table(description="Orders, one row per order.", unique_key="order_id")
def orders(ctx):
    return api.paginate(
        "/orders", paginator=sl.page_number("page"), select="$.items"
    )

The same holds for JSON you already have as bytes: yield sl.json(data) or sl.ndjson(data) and the bytes are parsed in Rust. See HTTP APIs for the full reference.

Describe the job, or yield the rows

You have	Yield	Where the work happens
A REST API that sl.http can describe	An sl.http job	Entirely in Rust: requests, pages, parsing, Arrow.
JSON or NDJSON as bytes, such as a response body or a file	sl.json(data) or sl.ndjson(data)	The bytes cross once; parsing and Arrow conversion run in Rust.
Arrow data: pyarrow, polars or pandas	The table, batch or DataFrame	Python builds it; it crosses as Arrow batches through the Arrow C interface, keeping its types.
Rows built in Python, such as from an SDK	Dicts, or lists of dicts	Python handles every row; SQLLoad types and converts them.

Pick the first row that fits. Describe the job when sl.http covers the API; yield your own rows or Arrow data when the source needs Python: an SDK, unusual authentication, or logic between requests.

The API at a glance

import sqlload as sl

# A source: the settings its tables share.
shop = sl.source("shop", write_strategy="merge")


# A table: one function, one table.
@shop.table(unique_key="order_id", cursor_column="updated_at")
def orders(ctx: sl.Context):
    """Orders from the shop, one row per order."""
    # ctx.cursor_value is where the last load ended.
    for page in fetch_orders(updated_after=ctx.cursor_value):
        # Dicts, Arrow data, JSON bytes or an sl.http job.
        yield page

Name	What it is
sl.source	A source: its connection, target and the settings its tables share.
@source.table	A table, from a function. Declaring it never runs it; the function runs when the table loads.
ctx	The run’s context: ctx.cursor_value and ctx.cursor_end for the cursor window, ctx.state for a logged position, ctx.client and ctx.connection.
What a function yields	Dicts, Arrow data, sl.json and sl.ndjson bytes, sl.http jobs and sl.checkpoint positions, in any mix.
sl.http	REST jobs that run entirely in Rust.

Setting up

List the folders holding them in sqlload.toml:

python_paths = ["sources"]
default_target = "local"

[connections.warehouse]
kind = "duckdb"
database = "shop.duckdb"

[targets.local]
connection = "warehouse"
loader_schema = "raw"

Every sqlload.source and @source.table found there joins plan, apply and load. Declaring one never runs it: the function runs only when its table loads.

import sqlload as sl

shop = sl.source("shop", write_strategy="merge")


@shop.table(unique_key="order_id", cursor_column="updated_at")
def orders(ctx: sl.Context):
    """Orders from the shop, one row per order."""
    yield [
        {"order_id": 1, "status": "paid", "total": 42.5,
         "updated_at": "2026-11-01T09:30:00Z"},
        {"order_id": 2, "status": "placed", "total": 18.0,
         "updated_at": "2026-11-02T14:05:00Z"},
    ]

$ sqlload load
…
└── shop__orders                                    merge (unique_key: order_id)
    ├── source  python  sources/shop.py (shop)
    ├── schema  columns inferred on first load
    ├── route   read as Arrow -> DuckDB takes Arrow
…
created  raw.shop__orders  6 columns
loaded   raw.shop__orders  2 rows  0.0s

The table loads into shop__orders: {source}__{table}, unless the source or table sets table_name. A @sl.table declared without a source keeps its own name. Table names are unique across the project.

Declaring

Function	Returns
sl.source(name, *, description=None, **settings)	A Source. settings are its connection, destination, schema and the defaults its tables share.
@source.table or @source.table(*, name=None, description=None, **settings)	A Table attached to the source.
@sl.table or @sl.table(...)	A standalone Table, with the target’s defaults.
@source.client	Declares a client the source’s tables share as ctx.client, built once per run.

A table’s description is description=, or else the function’s docstring; plan requires one. Unknown settings raise TypeError when the module is imported.

Settings

Setting	On	Meaning
write_strategy, unique_key, cursor_column, cursor_start, lookback, append_cursor_inclusive, batch_size	source or table	As in sqlload.toml.
contract, on_schema_change, auto_widen, type_change_rebuild	source or table	See Contracts.
allow_empty	source or table	Let replace empty a table when the function yields no rows.
state	source or table	"logged" to keep a position in ctx.state. See Logged state.
columns	table	Pinned column types, such as {"total": "decimal(18,2)"}.
flatten	table	Top-level objects to expand one level into <object>__<key> columns.
mask, exclude_columns, allow_columns	table (the last two also source)	As for other sources.
table_name, destination_path	source or table	Destination name template, and folder in a files target.
connection, destination, schema, mask_key, parallel_tables	source	Its connection, target, schema, HMAC key, and whether tables sharing a client may load at once.

What a function yields

A table function is a generator (or returns an iterable). It can yield, in any mix:

Yield	Becomes
A dict, or a list of dicts	Rows, typed like JSON.
Arrow data: a pyarrow table or batch, a polars or pandas DataFrame	Rows, keeping their Arrow types.
sl.json(data, path=None)	Rows from a JSON document: an array of objects (at a dotted path, such as "data.items"), or objects. Parsed in Rust.
sl.ndjson(data)	Rows from newline-delimited JSON. Parsed in Rust.
An sl.http job	Rows the job fetches, entirely in Rust. See HTTP APIs.
sl.checkpoint(position)	A logged position covering the rows yielded before it.

Types

Dict values are typed like JSON: numbers, booleans and text keep their types, and nested dicts and lists become JSON columns. Also:

Python value	Column type
Decimal	decimal, with room for at least 19 digits before the point. More than 38 digits in all become text; NaN and infinity, NULL.
date	date
naive datetime	timestamp_ntz
datetime with a UTC offset	timestamp_tz (a column mixing both becomes text)
Enum	Its value.
dataclass	A JSON object.
numpy numbers	Numbers.
Other values, including subclasses such as pandas.Timestamp	Text.

columns overrides inference. A first load that yields no rows creates nothing unless columns are declared. Inferred columns don’t change type by themselves; see Inferred types.

Flattening objects

Nested values stay one JSON column (VARIANT on Snowflake) unless the table lists top-level objects in flatten:

from datetime import date
from decimal import Decimal

import sqlload as sl

store = sl.source("store")


@store.table(
    description="Orders",
    flatten=["customer"],
    columns={"total": "decimal(18,2)"},
)
def store_orders(ctx):
    yield {
        "order_id": 7,
        "total": Decimal("12.50"),
        "placed_on": date(2026, 11, 1),
        "customer": {"country": "GB", "address": {"city": "Leeds"}},
        "line_totals": {"sku-14": 12.5},
    }

That loads these DuckDB columns: order_id BIGINT, total DECIMAL(18,2), placed_on DATE, customer__country VARCHAR, customer__address JSON, line_totals JSON.

Each listed object is expanded exactly one level. Objects not listed, such as line_totals, stay JSON; never flatten maps whose keys are data. Lists are never expanded. A missing or null object leaves its columns NULL, and new keys become new columns under on_schema_change. The load fails when a listed value arrives as a scalar or a list.

The context

A table function receives a sl.Context:

Attribute	Meaning
ctx.cursor_value	Where to read from: the destination’s maximum minus lookback, cursor_start, or --start-cursor. None reads everything.
ctx.cursor_end	Where the window ends, with batch_size or --end-cursor.
ctx.cursor_column, ctx.cursor_inclusive	The cursor column, and whether the read includes cursor_value.
ctx.is_reload	True under --reload.
ctx.state	The logged position, with state="logged". Assign it to move the position.
ctx.connection	The source’s connection settings, as a mapping. ctx.connection("name") gives another connection’s, except connections a target writes to.
ctx.client	The source’s @source.client, built on first use.
ctx.table, ctx.source, ctx.logger	The table and source names, and a logger.

Rows outside the cursor window are skipped with a warning, so a function that ignores ctx.cursor_value still loads correctly, only slower.

Table functions never get credentials of a connection a [targets] entry writes to: a source using such a connection fails at plan.

Tables of a source with a @source.client load one at a time, because clients are rarely thread-safe. Set parallel_tables=True on the source when its client is.

HTTP APIs

The full sl.http reference. Every job below runs in Rust, as described in Describe the job, or yield the rows.

[connections.shop]
kind = "http"
base_url = "https://api.example.com"
token = "${ENV:SHOP_TOKEN}"

import sqlload as sl

shop = sl.source("shop", connection="shop", write_strategy="merge")
api = sl.http(shop, auth=sl.bearer(), rate_limit="10/s", retries=5)


@shop.table(description="Orders, one row per order.", unique_key="order_id")
def orders(ctx):
    return api.paginate(
        "/orders", paginator=sl.page_number("page"), select="$.items"
    )


@shop.table(
    description="Order lines.", unique_key=["order_id", "line_no"]
)
def order_items(ctx):
    order_ids = api.paginate(
        "/orders", paginator=sl.page_number("page"), select="$.items[*].order_id"
    )
    return api.fan_out(
        "/orders/{order_id}/items",
        over=order_ids,
        concurrency=8,
        select="$.items",
        on_status={404: "skip"},
    )

$ sqlload load
…
created  raw.shop__order_items  5 columns
loaded   raw.shop__order_items  10 rows  1.2s
created  raw.shop__orders       5 columns
loaded   raw.shop__orders       5 rows  1.0s

✓ Completed  2 loaded, 15 rows, 1.2s

Function	Meaning
sl.http(source, *, auth=None, rate_limit=None, retries=5, headers=None, connection=None)	Requests to the source’s http connection. rate_limit ("10/s", "600/m", "1000/h") is shared by every job on the connection with the same limit.
api.paginate(path, *, paginator, params=None, select=None, on_status=None)	Every page of path. select is where the rows are: $, or a dotted path such as $.data.items.
api.fan_out(path, *, over, concurrency=8, paginator=None, params=None, select=None, on_status=None)	path once per key of over, up to concurrency at a time, while keys still arrive. over is a paginate whose select picks keys ($.items[*].id fills {id}), or a list of values.

Paginator	Next page
sl.single()	None: one request.
sl.next_link(pointer)	The URL at pointer in the body, such as "$.next".
sl.link_header()	The Link header’s rel="next".
sl.offset(param="offset", *, limit, limit_param="limit")	Counts rows by limit; ends at a short page.
sl.page_number(param="page", *, start=1)	Counts pages; ends at an empty page.
sl.cursor(pointer, param)	The token at pointer goes back as param.

Auth	Sends
sl.bearer()	Authorization: Bearer <token>, the connection’s token.
sl.api_key(header=...) or sl.api_key(query=...)	The token in that header or query parameter.
sl.basic()	Basic auth with the connection’s username and password.

• Every paginate names its paginator; nothing is guessed.
• Credentials go only to the connection’s own scheme, host and port, never to a next-page link elsewhere.
• Transport errors, 429 and 5xx are retried with backoff and jitter, honouring Retry-After. Another status fails, unless on_status maps it to "skip".
• The job is checked before any request. Errors name the request, the status and the start of the body, with the connection’s secrets replaced by ***.
• A job’s rows never come back into Python. Reading a row from a job fails (rule python_row_loop); filter rows downstream.

Logged state

Most tables need no state: their position is the data’s own cursor_column. A table whose position the data can’t carry, such as an API’s sync token, opts in with state="logged":

@shop.table(write_strategy="append", state="logged")
def events(ctx: sl.Context):
    """Shop events, read page by page with a sync token."""
    token = ctx.state
    while True:
        page = fetch_events(since=token)
        yield from page.items
        token = page.next_token or token
        yield sl.checkpoint(token)
        if not page.has_more:
            break

sl.checkpoint(position) takes any JSON value and covers exactly the rows yielded before it. sl.checkpoint(None) clears the position. See Logged state. To inspect or correct a logged position, see Logged state.

Other functions

Name	Meaning
sl.__version__	The package version.
sl.executable()	The path of the sqlload command installed with the package, else the one on PATH.