Python API

The sqlload package declares Python sources: tables whose rows a Python function yields. To set them up, and for HTTP APIs, see Load from Python and HTTP APIs.

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.

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. Python tables don’t take lookahead or lookback_from.
contract, on_schema_change, auto_widen, type_change_rebuild	source or table	See Contracts.
checks	source or table	Checks on the staged rows, such as {"unique": ["order_id"]}. See Checks before publish.
constant_columns	source or table	From 0.37.0 (not released yet). Columns with the same value in every row, such as {"tenant_id": "acme"} or {"shard": {"value": "7", "type": "int"}}; a table’s replace its source’s. A yielded field of the same name, ignoring case, fails the table before anything is written. See Several tenants in one table.
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, *, fields=None, keep_record=None)	Rows from a JSON document: an array of objects (at a dotted path, such as "data.items"), or objects. Parsed in Rust. See Fields for fields and keep_record.
sl.ndjson(data, *, fields=None, keep_record=None)	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. From 0.27.0, a list column pinned as array(T) loads as an array instead. 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. Dicts, sl.json bytes and Arrow structs flatten the same way, and a typed child of an Arrow struct keeps its type.

The load fails when a listed value arrives as a scalar or a list, or when a key’s column name is also a field of the rows or comes from flattening another object. plan refuses:

• flatten on tables that aren’t Python tables;
• a dotted path, such as customer.address: list customer, and its nested values stay JSON;
• an object that exclude_columns, mask or columns also names. After flattening no column carries the object’s own name, so name the <object>__<key> columns instead.

Turning flatten on for a table that already holds the object as a JSON column leaves that column NULL from then on, next to the new key columns.

Fields

From 0.27.0. Without fields, each key of a JSON row object loads as a column under its own name, as written: a key updated-at makes a column updated-at, which the destination quotes, so queries must quote it too. fields loads only the values it names, under the names it gives. It works on sl.json, sl.ndjson, and an sl.http job’s paginate and fan_out, all in Rust:

@shop.table(
    description="Accounts.",
    write_strategy="merge",
    unique_key="id",
    cursor_column="updated_at",
    columns={"updated_at": "timestamp_tz"},
)
def accounts(ctx):
    return api.paginate(
        "/accounts",
        paginator=sl.offset(limit=50),
        select="$.items",
        fields={
            "id": sl.field("$.id", required=True),
            "account_code": "$.code",
            "updated_at": "$.updated-at",
            "city": "$.address.city",
        },
        keep_record="raw_json",
    )

• A path is $ followed by keys, such as $.address.city. A key may hold any character but .. $ alone is the whole object.
• A missing key, or a path that runs through a value that isn’t an object, loads NULL. sl.field(path, required=True) fails the table instead, naming the row and the path.
• keep_record adds the whole row object as a JSON column. It needs fields, and its name can’t be a field’s. A table with a mask refuses it, because the record would carry the masked values.
• Types are inferred as for any JSON. Pin them with the table’s columns, which plan shows. columns, flatten and mask name the projected columns, not the source’s keys.
• A table with a cursor_column must load it, as a field, as keep_record, or as a column that flattening a field makes. A job whose fields leave it out fails before any request; sl.json and sl.ndjson data fails before it is parsed.
• A fan-out’s over picks keys with its select and takes no fields.

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. False when append reads after the destination’s maximum.
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

sl.http describes REST jobs that run entirely in Rust. For worked examples, see HTTP APIs in the guide. The connection is an http connection.

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, fields=None, keep_record=None, method="GET", json_body=None, key=None, cells=None)	Every page of path. select is where the rows are: $, or a dotted path such as $.data.items. fields and keep_record (from 0.27.0) pick and rename each row’s values; see Fields. method and json_body are from 0.36.0; see POST with a JSON body. From 0.37.0 (not released yet), cells= turns each row’s list of cells into columns; see Rows of cells.
api.fan_out(path, *, over, concurrency=8, paginator=None, params=None, select=None, on_status=None, fields=None, keep_record=None, method="GET", json_body=None, key=None, cells=None)	path once per key of over, up to concurrency at a time, while keys still arrive. fields and keep_record are from 0.27.0. over is a paginate whose select picks keys ($.items[*].id fills {id}), or a list: a value fills the path’s one placeholder, and a dict fills placeholders by name. From 0.37.0 (not released yet), over can also be another fan_out; see Multi-level fan-out. method and json_body are from 0.36.0, as for paginate.

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", location="query")	Counts rows by limit; ends at a short page. location is from 0.36.0.
sl.page_number(param="page", *, start=1, location="query")	Counts pages; ends at an empty page. location is from 0.36.0.
sl.cursor(pointer, param, *, location="query")	The token at pointer goes back as param. location is from 0.36.0.

Cells	Columns
sl.cells(pointer="$.cells", *, key="columnId", value="$.value", names_from="$.columns", name_key="title", id_key="id", header_row=False)	From 0.37.0 (not released yet). Each row’s cells at pointer, one column per column the response names. See Rows of cells.

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.
sl.oauth2_client_credentials(token_url, *, scope=None, client_auth="basic")	From 0.27.0. Authorization: Bearer <access token>, with the token token_url issues for the connection’s client_id and client_secret. See OAuth2.
sl.oauth2_refresh_token(token_url, *, scope=None, client_auth="basic")	From 0.27.0. The same, with the token issued for the connection’s refresh_token. The connection’s client_id and client_secret go along when it has them.
sl.oauth2_password(token_url, *, scope=None, client_auth="basic")	From 0.36.0. The same, with the token issued for the connection’s username and password (the resource-owner password grant). The connection’s client_id and client_secret go along when it has them.

• 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. The one exception is OAuth2: the client secret and refresh token go to its token_url.
• 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.

Multi-level fan-out

From 0.37.0 (not released yet). A fan_out whose select picks keys can be the over of another fan_out, as many levels deep as they nest: list → detail → sub-detail. Each key keeps the fields of the keys above it, so a third level’s path can use {order_id} as well as {item_id}. When a level’s field has the same name as one above it (every level calls it id), name it apart with key=, as in select="$.items[*].id", key="item_id".

@shop.table(description="Order notes.", write_strategy="append", cursor_column="order_id")
def notes(ctx):
    orders = api.paginate("/orders", paginator=sl.page_number("p"),
                          params={"after": ctx.cursor_value}, select="$.items[*].order_id")
    items = api.fan_out("/orders/{order_id}/items", over=orders, select="$.items[*].item_id")
    return api.fan_out("/orders/{order_id}/items/{item_id}/notes", over=items,
                       paginator=sl.page_number("page"), select="$.items")

• Every level has its own paginator, concurrency, on_status and, through the sl.http it comes from, its own retries and rate limit. Up to one more request than the sum of the fan-out levels’ concurrency is in flight at once (the top paginate’s); a shared rate_limit caps them.
• An incremental cursor goes on the top level, as above; lower levels only see the keys it picks.
• Only the last level’s rows load. A level that should be a table of its own is a table function whose job stops at that level, so the levels above it are requested again for that table.
• Rows load as the API returns them: a key’s fields are not added to them, so select an ID the response repeats, such as order_id above.
• An error names the keys that led to it, such as /orders order_id=5 → /orders/5/items item_id=51 → GET .../notes?page=2: HTTP 500.
• plan checks every level before any request: each placeholder must come from a level above.

Rows of cells

From 0.37.0 (not released yet). Some APIs return each row as a list of cells, such as [{"columnId": 7001, "value": "Bolt"}], and list the column names separately (Smartsheet), or name them in the first row (Google’s grid data). cells=sl.cells(...) on paginate or fan_out turns each row into one column per named column, in Rust, before the rows load:

@sheets.table(description="Parts.")
def parts(ctx):
    return api.paginate("/sheets/123", paginator=sl.page_number("page"), select="$.rows",
                        cells=sl.cells(value=["$.value", "$.displayValue"]))

• A cell’s column is its key field (columnId), matched to the id_key (id) of the columns listed at names_from ($.columns), named by their name_key (title). These defaults fit Smartsheet.

• key=None matches cells to columns by position. header_row=True takes the names from the first row’s cells instead of names_from, and that row doesn’t load. For Google’s grid data:

  api.paginate("/spreadsheets/ID", paginator=sl.single(), select="$.sheets[0].data[0].rowData",
               cells=sl.cells("$.values", key=None, header_row=True,
                              value=["$.effectiveValue.stringValue", "$.effectiveValue.numberValue",
                                     "$.effectiveValue.boolValue"]))

• value is where each cell’s value is, or a list of places tried in turn; the first present and not null wins. Values are typed like any other JSON value: a number loads as a number, true as a boolean. Paths in select and here may index a list, as in $.sheets[0].

• A row with no cell for a column loads NULL in it. The row’s other fields, such as its id or modifiedAt, load beside the cells’ columns; one with the name of a cells’ column fails the load.

• A cell whose column the response doesn’t name fails the load, naming its id, unless it has no value (Google returns formatted empty cells past the last column). Two cells for one column in a row fail it too, as does a page whose rows have no field at pointer. Names are read again from every page that lists them; a page that doesn’t keeps the names of the page before it.

• header_row reads the first row of a paginate’s first page. In a fan_out, that is each key’s first page, which suits one sheet per key; a fan-out that splits one sheet into ranges would lose each range’s first row, so list the names with names_from instead.

• A column renamed in the source is a new column to the destination, so on_schema_change decides: by default it is added, and earlier rows hold NULL in it.

• fields= picks from the pivoted columns, by their names ({"qty": "$.Qty"}), and keep_record keeps the pivoted row, not the cells as the API sent them.

POST with a JSON body

From 0.36.0. method="POST" sends json_body with every request, as Content-Type: application/json; GET stays the default, and its requests don’t change. With location="body", sl.offset, sl.page_number and sl.cursor put their values in the body instead of the query string, and param (and limit_param) are paths such as "$.page.cursor". Each request starts from a fresh copy of json_body, so one page’s values never leak into the next, and a retry resends the same body. A POST without json_body starts from {}.

To send the incremental cursor in the body, put ctx.cursor_value in json_body, as you would in params. Dates, times and decimals go as text (ISO text for dates and times); wrap the value in int() or float() if the API wants a JSON number.

@shop.table(description="Orders.", write_strategy="append", cursor_column="updated_at")
def orders(ctx):
    return api.paginate(
        "/orders/search",
        method="POST",
        json_body={"filter": {"updated_after": ctx.cursor_value}},
        paginator=sl.cursor("$.meta.next", "$.page.cursor", location="body"),
        select="$.items",
    )

json_body on a GET, location="body" without method="POST", or a body path that json_body already holds as a non-object (such as $.page.cursor with {"page": 1}) fails before any request. A POST follows only 307 and 308 redirects, which keep the method and body; a 301, 302 or 303 would resend it as a GET without its body, so it fails instead. sl.next_link and sl.link_header resend the same body to the next page, and fail if that page is on another origin, so json_body never leaves the connection’s origin.

OAuth2

From 0.27.0. sl.oauth2_client_credentials, sl.oauth2_refresh_token and (from 0.36.0) sl.oauth2_password fetch short-lived access tokens from token_url. See OAuth2 in the guide for an example.

• The token is fetched once per run and shared by every job on the connection with the same auth. It is renewed before it expires (by expires_in), and once more when the API answers 401; the request is then retried with the new token. When on_status maps 401 to "skip", a 401 renews the token only if it is over 30 seconds old; otherwise the request is skipped.
• client_auth="basic" sends client_id and client_secret to the token URL with HTTP basic auth, the method RFC 6749 requires servers to support. As the RFC asks, both are form-encoded first, so a secret with characters such as + or / reaches the server encoded. Use "post" for a server that rejects that, or that wants them as form fields.
• The client secret, refresh token and password go only to token_url, which must be https (http only on localhost). Access tokens go only to the connection’s own origin, like every other credential.
• Token errors name the token URL, the status and the server’s error and error_description. The client secret, refresh token, password and every access token are shown as ***.
• When the server rotates the refresh token, SQLLoad uses the new one for the rest of the run but does not save it, so each run starts from the connection’s refresh_token. An API that revokes the old refresh token on rotation therefore fails on the next run; use client credentials there when the API offers them.
• From 0.36.0, sl.oauth2_password posts grant_type=password with the connection’s username and password. When the response includes a refresh_token, later renewals in the run use it; when it doesn’t, or the server refuses it, SQLLoad asks with the password again. The password is shown as *** like the other secrets, and wrong credentials end the run with exit 1 (not retryable) and the server’s error.

[connections.shop]
kind = "http"
base_url = "https://api.example.com/v1"
username = "loader"
password = "${ENV:SHOP_PASSWORD}"
client_id = "sqlload"

api = sl.http(shop, auth=sl.oauth2_password("https://auth.example.com/oauth/token", client_auth="post"))

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 sqlload 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.