Platforms

base

PlatformAdapter Protocol: the contract every platform integration implements.

PlatformAdapter

Bases: Protocol

Executes SQL against a data platform; returns rows + schema + latency.

Required behavior

• On success: return ExecutionResult with rows populated and schema_ populated — a TypedSchema (each Column.type a SqlType built from the driver's native type string in the adapter's static dialect), or an UntypedSchema (names only) when the driver reports no result-column types — with non-negative latency_seconds and error is None.
• On query failure: return ExecutionResult with rows=[], schema_=None, an error describing the failure, and non-negative latency_seconds. Do NOT raise.
• latency_seconds measures wall-clock time spent inside execute().

execute

execute(sql: str) -> ExecutionResult

Execute one SQL statement and return its structured result.

cancel

cancel() -> None

Abort the query currently executing on this connection, if any.

Must be safe to call from another thread while execute is blocked. A no-op when no query is running, and best-effort: it must not raise — a cancellation that fails or arrives late is non-fatal.

close

close() -> None

Release the underlying connection/resources.

TypeResolvingAdapter

Bases: Protocol

Capability for backends whose execute().schema_ reports types with unresolved parameters.

Implemented only by adapters whose driver drops type parameters (e.g. a DECIMAL scale or an ARRAY element type); precise backends do not implement it. Both methods must be pure (no I/O): one produces the SQL that probes for precise types, the other interprets that probe's result rows. The caller runs the probe.

type_probe_sql

type_probe_sql(sql: str) -> str

Build the probe statement that recovers precise types for sql's projection.

Parameters:

Name	Type	Description	Default
sql	str	The statement whose projected column types to resolve.	required

Returns:

Type	Description
str	A probe statement for the caller to execute; its rows feed types_from_probe.

types_from_probe

types_from_probe(
    rows: list[dict[str, Any]],
) -> list[SqlType] | ExecutionError

Parse the probe's result rows into precise SqlTypes, one per projected column.

They must be returned in the order the query projects its columns.

Parameters:

Name	Type	Description	Default
rows	list[dict[str, Any]]	The rows returned by executing type_probe_sql, in projection order.	required

Returns:

Type	Description
list[SqlType] | ExecutionError	One precise SqlType per projected column, in order, or an ExecutionError when
list[SqlType] | ExecutionError	the rows cannot yield types.

execute_within_budget

execute_within_budget(
    adapter: PlatformAdapter,
    sql: str,
    max_seconds: float | None,
) -> ExecutionResult

Execute sql on adapter, cancelling it if it exceeds max_seconds.

With max_seconds unset, runs adapter.execute directly. Otherwise runs it on a worker thread and waits up to max_seconds; if the query is still running, calls adapter.cancel() to abort it and returns an ExecutionResult.error describing the overrun. The cancelled query is awaited so its connection is released before returning.

Parameters:

Name	Type	Description	Default
adapter	PlatformAdapter	The platform adapter to execute against.	required
sql	str	The SQL statement to execute.	required
max_seconds	float | None	Wall-clock ceiling for the query, or None for no limit.	required

Returns:

Type	Description
ExecutionResult	The adapter's ExecutionResult if the query finishes within budget, otherwise an
ExecutionResult	ExecutionResult with error set and latency_seconds measuring the elapsed time.

execution_error

execution_error(
    exc: Exception,
    kind: ExecutionErrorKind = "query_failed",
) -> ExecutionError

Translate a driver exception into a typed ExecutionError, preserving structured detail.

Reads whatever structured attributes the exception happens to expose — SQLSTATE, an error condition/class, message parameters — by duck-typing, so no driver package is imported or enumerated and no driver message is parsed. The live exception is kept as cause for debugging; message falls back to the exception's class name when its string form is empty.

Parameters:

Name	Type	Description	Default
exc	Exception	The driver exception to translate.	required
kind	ExecutionErrorKind	The ExecutionError classifier to assign.	'query_failed'

Returns:

Type	Description
ExecutionError	An ExecutionError carrying kind, a non-empty message, any recovered structured
ExecutionError	fields, and exc as cause.

rows_or_error

rows_or_error(
    columns: list[Column],
    rows_raw: list[tuple[Any, ...]],
    latency_seconds: float,
) -> ExecutionResult

Build a successful typed ExecutionResult, or an error one if column names collide.

Parameters:

Name	Type	Description	Default
columns	list[Column]	The result-set columns, in order, as reported by the driver.	required
rows_raw	list[tuple[Any, ...]]	The positional row tuples, aligned with columns.	required
latency_seconds	float	Wall-clock time spent executing the query.	required

Returns:

Type	Description
ExecutionResult	An ExecutionResult with name-keyed rows and a TypedSchema, or error set (and no
ExecutionResult	rows or schema) when one or more column names are duplicated.

untyped_rows_or_error

untyped_rows_or_error(
    names: list[str],
    rows_raw: list[tuple[Any, ...]],
    latency_seconds: float,
) -> ExecutionResult

Build a successful untyped ExecutionResult, or an error one if column names collide.

For engines whose driver reports no result-column types (e.g. SQLite): the schema carries column names only.

Parameters:

Name	Type	Description	Default
names	list[str]	The result-column names, in order, as reported by the driver.	required
rows_raw	list[tuple[Any, ...]]	The positional row tuples, aligned with names.	required
latency_seconds	float	Wall-clock time spent executing the query.	required

Returns:

Type	Description
ExecutionResult	An ExecutionResult with name-keyed rows and an UntypedSchema, or error set (and
ExecutionResult	no rows or schema) when one or more column names are duplicated.

registry

Platform-reference builders and PlatformRef -> live PlatformAdapter resolution.

duckdb_platform

duckdb_platform(
    name: str, path: str = ":memory:"
) -> PlatformRef

Build a PlatformRef for an in-process DuckDB database.

Parameters:

Name	Type	Description	Default
name	str	A unique name identifying this platform connection.	required
path	str	The DuckDB database path. Defaults to :memory: (in-process).	':memory:'

Returns:

Type	Description
PlatformRef	A serializable PlatformRef for the DuckDB database. Building the ref needs no
PlatformRef	driver.

sqlite_platform

sqlite_platform(
    name: str, path: str = ":memory:"
) -> PlatformRef

Build a PlatformRef for an in-process SQLite database.

Parameters:

Name	Type	Description	Default
name	str	A unique name identifying this platform connection.	required
path	str	The SQLite database path. Defaults to :memory: (in-process).	':memory:'

Returns:

Type	Description
PlatformRef	A serializable PlatformRef for the SQLite database. Building the ref needs no driver
PlatformRef	(SQLite ships with the standard library).

postgres_platform

postgres_platform(
    name: str, conninfo: str = ""
) -> PlatformRef

Build a PlatformRef for a PostgreSQL database.

Parameters:

Name	Type	Description	Default
name	str	A unique name identifying this platform connection.	required
conninfo	str	A libpq connection string (keyword/value or postgresql:// URI). Empty uses libpq defaults / PG* env vars.	''

Returns:

Type	Description
PlatformRef	A serializable PlatformRef for the PostgreSQL database. Building the ref needs no
PlatformRef	driver.

databricks_platform

databricks_platform(
    name: str,
    *,
    server_hostname: str,
    http_path: str,
    catalog: str | None = None,
    schema: str | None = None,
) -> PlatformRef

Build a PlatformRef for a Databricks SQL Warehouse.

Holds only non-secret connection details; credentials are not included here.

Parameters:

Name	Type	Description	Default
name	str	A unique name identifying this platform connection.	required
server_hostname	str	The workspace hostname (no scheme), e.g. dbc-xxxx.cloud.databricks.com.	required
http_path	str	The SQL Warehouse HTTP path, e.g. /sql/1.0/warehouses/<id>.	required
catalog	str | None	The default catalog, or None to leave the session default.	None
schema	str | None	The default schema, or None to leave the session default.	None

Returns:

Type	Description
PlatformRef	A serializable PlatformRef for the Databricks warehouse. Building the ref needs no
PlatformRef	driver.

snowflake_platform

snowflake_platform(
    name: str,
    *,
    account: str,
    user: str | None = None,
    warehouse: str | None = None,
    role: str | None = None,
    database: str | None = None,
    schema: str | None = None,
    authenticator: str | None = None,
    workload_identity_provider: str | None = None,
) -> PlatformRef

Build a PlatformRef for a Snowflake account.

Holds only non-secret connection details; credentials are not included here.

Parameters:

Name	Type	Description	Default
name	str	A unique name identifying this platform connection.	required
account	str	The Snowflake account identifier.	required
user	str | None	The Snowflake user name, or None to rely on authenticator.	None
warehouse	str | None	The default warehouse, or None to leave the session default.	None
role	str | None	The default role, or None to leave the session default.	None
database	str | None	The default database, or None to leave the session default.	None
schema	str | None	The default schema, or None to leave the session default.	None
authenticator	str | None	The authenticator to use (e.g. "externalbrowser", "oauth", "workload_identity"), or None for the connector's default.	None
workload_identity_provider	str | None	The workload identity provider (e.g. "OIDC") when authenticator is "workload_identity", or None otherwise.	None

Returns:

Type	Description
PlatformRef	A serializable PlatformRef for the Snowflake account. Building the ref needs no
PlatformRef	driver.

resolve

resolve(ref: PlatformRef) -> PlatformAdapter

Return a live adapter for ref, building and caching one on first use.

Reuses the cached adapter on subsequent calls for the same ref.name. An unsupported kind is unrepresentable — PlatformRef validation rejects it before resolution.

Parameters:

Name	Type	Description	Default
ref	PlatformRef	The platform reference to resolve.	required

Returns:

Type	Description
PlatformAdapter	The live PlatformAdapter for ref, cached and reused across calls.

Raises:

Type	Description
ValueError	If ref.name is already bound to a different configuration.

close_all

close_all() -> None

Close every cached adapter and clear the cache (idempotent; no-op when empty).

duckdb

DuckDBAdapter: in-process DuckDB execution backend.

DuckDBAdapter

DuckDBAdapter(database: str = ':memory:')

Executes SQL against an in-process DuckDB database.

Open a DuckDB connection to database (default :memory:).

cancel

cancel() -> None

Interrupt the query currently executing on this connection.

Safe to call from another thread while execute is blocked, and a no-op when no query is running. The interrupted execute raises duckdb.InterruptException, which it surfaces as ExecutionResult.error like any other query failure.

close

close() -> None

Release the underlying DuckDB connection (file handle / WAL lock).

Explicit close matters on Windows, where WAL locks make implicit cleanup unreliable.

execute

execute(sql: str) -> ExecutionResult

Execute one SQL statement against the database.

Parameters:

Name	Type	Description	Default
sql	str	The SQL statement to execute.	required

Returns:

Type	Description
ExecutionResult	An ExecutionResult with the returned rows, schema, and latency. Query
ExecutionResult	failures are returned as ExecutionResult.error rather than raised.

postgres

PostgresAdapter: PostgreSQL execution backend over psycopg (v3).

PostgresAdapter

PostgresAdapter(conninfo: str = '')

Executes SQL against a PostgreSQL database via psycopg (v3).

Open a psycopg connection.

conninfo is a libpq connection string — keyword/value ("host=... port=... user=... password=... dbname=...") or a postgresql:// URI. Empty uses libpq defaults / PG* env vars.

cancel

cancel() -> None

Cancel the query currently executing on this connection.

Sends a libpq cancel request over a separate channel, so it is safe to call from another thread while execute is blocked, and a harmless no-op when no query is running. The cancelled execute raises psycopg.errors.QueryCanceled, surfaced as ExecutionResult.error. Cancellation failures are swallowed — they are non-fatal.

close

close() -> None

Release the underlying psycopg connection.

execute

execute(sql: str) -> ExecutionResult

Execute one SQL statement against the database.

Parameters:

Name	Type	Description	Default
sql	str	The SQL statement to execute.	required

Returns:

Type	Description
ExecutionResult	An ExecutionResult with the returned rows, schema, and latency. Query
ExecutionResult	failures are returned as ExecutionResult.error rather than raised.

databricks

DatabricksAdapter: Databricks SQL Warehouse execution backend over databricks-sql-connector.

DatabricksAdapter

DatabricksAdapter(
    *,
    server_hostname: str,
    http_path: str,
    catalog: str | None = None,
    schema: str | None = None,
)

Executes SQL against a Databricks SQL Warehouse via databricks-sql-connector.

Open a Databricks SQL connection.

Credentials are not passed here.

Parameters:

Name	Type	Description	Default
server_hostname	str	The workspace hostname (no scheme), e.g. dbc-xxxx.cloud.databricks.com.	required
http_path	str	The SQL Warehouse HTTP path, e.g. /sql/1.0/warehouses/<id>.	required
catalog	str | None	The default catalog, or None to leave the session default.	None
schema	str | None	The default schema, or None to leave the session default.	None

cancel

cancel() -> None

Cancel the query currently executing on this connection, if any.

Safe to call from another thread while execute is blocked; best-effort, so all failures are swallowed.

close

close() -> None

Release the underlying Databricks connection.

execute

execute(sql: str) -> ExecutionResult

Execute one SQL statement against the warehouse.

Parameters:

Name	Type	Description	Default
sql	str	The SQL statement to execute.	required

Returns:

Type	Description
ExecutionResult	An ExecutionResult with the returned rows, schema, and latency. Query
ExecutionResult	failures are returned as ExecutionResult.error rather than raised.

type_probe_sql staticmethod

type_probe_sql(sql: str) -> str

Build the DESCRIBE QUERY probe that recovers precise types for sql.

cursor.description drops parametric type parameters (a DECIMAL scale, an ARRAY element type); DESCRIBE QUERY reports the precise data_type per column without running the query.

Parameters:

Name	Type	Description	Default
sql	str	The statement whose projected column types to resolve.	required

Returns:

Type	Description
str	A DESCRIBE QUERY statement wrapping sql (trailing ; stripped, which it rejects).

types_from_probe staticmethod

types_from_probe(
    rows: list[dict[str, Any]],
) -> list[SqlType] | ExecutionError

Parse DESCRIBE QUERY rows into precise SqlTypes, in projection order.

Parameters:

Name	Type	Description	Default
rows	list[dict[str, Any]]	The rows returned by executing type_probe_sql, in projection order.	required

Returns:

Type	Description
list[SqlType] | ExecutionError	One precise SqlType per projected column, in order, or an ExecutionError when
list[SqlType] | ExecutionError	the probe returns no rows or a row without a data_type.