etlantic-sql API

Status: Available in ETLantic 0.41.0. SQL plugin (SQLite + PostgreSQL). Install narrative: package README. Hub: Optional packages API.

Setup

pip install 'etlantic-sql==0.41.0'

import etlantic_sql
print(etlantic_sql.__version__)

Failure modes

Topic	Behavior
Engine URL / dialect	PMEXEC433 empty engine; MERGE PostgreSQL-only

Public API

etlantic_sql

etlantic-sql — PostgreSQL / SQLite Tier A reference SQL execution plugin.

async_executor

Async SQLAlchemy execution path for the reference SQL plugin.

AsyncSqlExecutor

AsyncSqlExecutor(*, engine: Any, dialect: str, rows_fetched_counter: list[int], bound_params: MutableMapping[str, dict[str, Any]], staging_tables: list[str])

Run compiled statements inside an async transaction.

catalog

Catalog / information_schema inspection (metadata only, no row reads).

create_table_from_model

create_table_from_model(engine: Engine, model: Any, *, dialect: str, checkfirst: bool = True) -> dict[str, Any]

Create a physical table from SQLModel / SQLAlchemy Table metadata.

Never reads source rows. Returns secret-free catalog metadata.

validate_primary_keys

validate_primary_keys(engine: Engine, relation: RelationRef, *, expected_keys: Sequence[str], dialect: str) -> dict[str, Any]

Validate primary-key columns against catalog metadata; fail closed.

pk_validation_diagnostics

pk_validation_diagnostics(result: Mapping) -> list[Diagnostic]

Convert validate_primary_keys result into Diagnostic objects.

compiler

Compile portable SQL IR into parameterized dialect SQL.

SqlCompiler

SqlCompiler(*, dialect: str, supports_merge: bool)

IR → parameterized SQL text (values never interpolated).

connectors

PostgreSQL source/sink connectors for etlantic-sql (fake/sqlite CI path).

FakePostgresConnection dataclass

FakePostgresConnection(conn: Connection = (lambda: sqlite3.connect(':memory:'))(), _pending_ops: dict[str, dict[str, Any]] = dict(), _committed_ops: dict[str, dict[str, Any]] = dict(), _query_seq: int = 0)

SQLite-backed transactional fake for CI without a live Postgres.

Mirrors autocommit-off semantics: DML is held until commit/rollback. Pending query/op ids move to committed only on commit(); rollback() clears pending so reconciliation cannot treat aborted work as committed.

PostgresSourceConnector dataclass

PostgresSourceConnector(connection: FakePostgresConnection = FakePostgresConnection())

Bounded reads against PostgreSQL (sqlite fake in CI).

PostgresSinkConnector dataclass

PostgresSinkConnector(connection: FakePostgresConnection = FakePostgresConnection(), _sessions: dict[str, dict[str, Any]] = dict())

Transactional sink with commit/rollback and query-id evidence.

create_source

create_source() -> PostgresSourceConnector

Entry-point factory for etlantic.source_connectors (postgresql).

create_sink

create_sink() -> PostgresSinkConnector

Entry-point factory for etlantic.sink_connectors (postgresql).

create_storage

create_storage() -> PostgresStorageConnector

Entry-point factory for etlantic.storage_connectors (postgresql).

dialect_postgresql

Dialect helpers for identifier quoting and dialect detection.

detect_dialect

detect_dialect(url: str) -> str

Return the normalized dialect name for url.

dialect_info

dialect_info(url: str) -> DialectInfo

Return full dialect tier metadata for url.

quote_identifier

quote_identifier(name: str, *, dialect: str = 'postgresql') -> str

Quote a validated SQL identifier (double-quotes for PG and SQLite).

dialect_tiers

SQL dialect support tiers for the reference etlantic-sql plugin.

Tier A dialects are exercised in CI (SQLite + PostgreSQL). Tier B dialects are detected and capability-gated — planning must fail closed for features the plugin does not truthfully implement.

DialectInfo dataclass

DialectInfo(name: str, tier: str, scheme: str)

Normalized dialect identity and support tier.

detect_dialect_info

detect_dialect_info(url: str) -> DialectInfo

Classify a SQLAlchemy URL into a dialect name and support tier.

assert_tier_a_or_raise

assert_tier_a_or_raise(info: DialectInfo, *, feature: str) -> None

Fail closed when a Tier A-only feature is requested on another dialect.

executor

Connection / transaction execution for compiled SQL.

SqlExecutor

SqlExecutor(*, engine: Engine, dialect: str, rows_fetched_counter: list[int], bound_params: MutableMapping[str, dict[str, Any]], staging_tables: list[str])

Run compiled statements inside a transaction.

materialize_temp

materialize_temp(compiler: SqlCompiler, query: SqlQuery, *, temp_name: str, params: Mapping[str, Any], context: SqlExecutionContext, seal: Any) -> SqlExecutionResult

Materialize as a durable run-scoped table (visible across pool checkouts).

publish_replace

publish_replace(*, target: RelationRef, staging: RelationRef, compiler: SqlCompiler, context: SqlExecutionContext) -> SqlExecutionResult

Swap staging into target without dropping the live table first.

frame

In-memory relation handle used by portable SQL conformance.

SqlRelationFrame dataclass

SqlRelationFrame(rows: list[dict[str, Any]] = list(), name: str = '')

Row-backed relation used as a portable SQL compiler input/output.

lowering

DTCS → etlantic.sql/1 lowering for the portable SQL compiler.

apply_action_to_query

apply_action_to_query(source: RelationRef, columns: list[str], action: dict[str, Any], *, parameters: dict[str, Any], relations: dict[str, RelationRef], relation_columns: dict[str, list[str]]) -> tuple[SqlQuery, list[str]]

Lower one action into a SqlQuery over source.

Returns the query and the resulting output column names.

actions

Lower DTCS kernel/relational actions onto SqlQuery IR.

apply_action_to_query

apply_action_to_query(source: RelationRef, columns: list[str], action: dict[str, Any], *, parameters: dict[str, Any], relations: dict[str, RelationRef], relation_columns: dict[str, list[str]]) -> tuple[SqlQuery, list[str]]

Lower one action into a SqlQuery over source.

Returns the query and the resulting output column names.

expr

Lower DTCS expression nodes to etlantic.sql IR expressions.

plugin

SQLAlchemy-backed SQL plugin (PostgreSQL + SQLite Tier A reference).

PostgresSqlPlugin

PostgresSqlPlugin(*, url: str | None = None)

Reference SQL plugin (Tier A: PostgreSQL + SQLite via SQLAlchemy Core).

get_engine

get_engine() -> Engine

Return the shared SQLAlchemy engine for this plugin instance.

execute_async async

execute_async(compiled: Sequence[CompiledSql], *, params: Mapping[str, Any], context: SqlExecutionContext, fetch: bool = False) -> SqlExecutionResult

AsyncEngine execution path (requires async driver URL + extras).

create_table_from_model

create_table_from_model(model: Any, *, checkfirst: bool = True) -> dict[str, Any]

Create a table from a SQLModel / SQLAlchemy Table metadata object.

validate_primary_keys

validate_primary_keys(relation: RelationRef, *, expected_keys: Sequence[str], context: SqlExecutionContext | None = None) -> dict[str, Any]

Validate that relation declares the expected primary-key columns.

cleanup_staging

cleanup_staging() -> None

Drop run-scoped durable staging tables.

create_plugin

create_plugin() -> PostgresSqlPlugin

Entry-point factory for etlantic.sql_plugins.

transform_compiler

SQL portable transform compiler (kernel + relational claims).

SqlTransformCompiler

SqlTransformCompiler()

Compile dtcs.transform-plan/2 kernel+relational IR to SQL IR.

create_transform_compiler

create_transform_compiler() -> SqlTransformCompiler

Entry-point factory for etlantic.transform_compilers.

writes

Write-intent helpers for the reference SQL plugin.