etlantic-sql

SQLite and PostgreSQL reference SQL execution plugin for ETLantic 0.43. Install when pipelines need Profile(sql_engine="sql"), SQL→SQL fusion, or Experimental postgresql source/sink/storage connectors. Keep the pin matched to core.

Note: This plugin and ETLantic core use Beta classifiers for documented single-tenant pilots. Classifiers are not an enterprise SLA.

Install

pip install 'etlantic-sql==0.57.2'
# pip install 'etlantic==0.57.2'
export ETLANTIC_SQL_URL=postgresql+psycopg://user:pass@localhost:5432/etlantic
# Or use SQLite:
# export ETLANTIC_SQL_URL=sqlite+pysqlite:///:memory:

Uses SQLAlchemy Core. Driver dependencies stay out of etlantic core.

Wiring

from etlantic import Profile

Profile(name="sql-prod", sql_engine="sql")

The etlantic.sql_plugins entry point named sql registers etlantic_sql:create_plugin. Profiles select it with sql_engine="sql"; keep connection URLs in environment-backed configuration or secret providers, not in plans.

Prefer @Transformation.portable with portable_transform_policy="require"; the compiler lowers the qualified baseline to typed, parameterized SQL. Register @Transformation.implementation("sql") only when dialect-specific behavior outside that baseline is required. Native handlers take RelationRef inputs and return SQL query handles (not fetched rows); they are tied to SQL and are not eligible for adaptive execution.

Capabilities

• SQL→SQL fusion without intermediate Python row fetch
• Durable run-scoped staging tables (not session TEMP)
• Insert-select / CTAS-style publication
• Fail-closed planning when required capabilities are missing

SQLite and PostgreSQL are Tier A in 0.33. MERGE / upsert is advertised only for PostgreSQL (sql_merge=True, INSERT … ON CONFLICT); SQLite remains sql_merge=False and fails closed when merge is required.

Inference target mode reference (0.55)

SQLiteTableTarget is a separate, explicit reference adapter for an existing SQLite table. It inspects declared columns, primary keys, and configured partition columns without reading rows. Its transactional write_records method executes append, overwrite, merge, upsert, and partition replacement only when the table advertises the requested mode and its revision still matches. This adapter does not change the SQL engine capability setting above.

import sqlite3
import etlantic as etl
from etlantic_sql import SQLiteTableTarget

connection = sqlite3.connect("orders.db")
target = SQLiteTableTarget(
    connection, "orders", identity="orders", partitions=("order_day",)
)
observation = etl.inspect_target(target)

PostgreSQL connectors (0.57 implementation; Experimental)

The postgresql source, sink and storage entry points use SQLAlchemy and psycopg against a live PostgreSQL database. They require a runtime SecretValue containing the connection URL, or a worker-level ETLANTIC_SQL_URL. Put table and mode options in the public asset config; do not put a URL or credential in that config. The source reads one repeatable read snapshot, with configurable row_limit, batch_size and max_bytes (bounded to 100,000 rows, 10,000 records per batch and 256 MiB). Schema inspection reads catalog metadata and PostgreSQL's row estimate without creating or changing a target.

The sink supports append, replace/overwrite, and upsert. Upsert requires key_columns that match a primary or unique constraint. Commit recovery relies on a provisioned durable effect ledger. Provision it once through a database administrator or migration before enabling sink writes:

CREATE TABLE public.etlantic_connector_effects (
    effect_id text PRIMARY KEY,
    intent_fingerprint text NOT NULL,
    publication_id text NOT NULL UNIQUE,
    row_count bigint NOT NULL,
    committed_at timestamptz NOT NULL DEFAULT now()
);

The worker takes a transaction-scoped advisory lock for each stable run/node effect ID, changes the target and inserts the ledger row in the same transaction. Reconciliation consults that row; an unavailable ledger returns unknown. An effect ID bound to different write intent fails closed. The SQLite-backed fake remains available as FakePostgresConnection for fast connector unit tests; it is not registered as the PostgreSQL provider.

Managed repair and backfill use bounded partition operations when both PostgreSQL bindings configure partition_column and declare source.partitioned or write.partition_replace plus idempotency as required capabilities. Partition IDs are opaque strings matched to the configured column's text representation. Source reads filter to those values; sink publication deletes and replaces only those values in the same transaction as its effect-ledger row. Empty output clears the selected partitions. Rows outside the accepted selector are rejected before commit. Bindings without the column configuration cannot admit repair or backfill.

During managed execution, PostgreSQL source and sink connectors also provide opaque resource identities to the worker. Before opening a sink write session, the runtime compares source and target identities, including the live server address and normalized schema/table. A matching identity, or a same-provider transfer whose identities cannot be verified, fails with PMEXEC435 before target mutation. The comparison tokens use a worker-process key and stay in memory; credentials and raw database/resource names are not written to plans, receipts or reports.

Managed connector action workers (0.57)

The public create_action_handlers(resolve_engine) factory registers PostgreSQL connection tests, bounded schema inspection, live catalog lookup, and explicit provisioning/cleanup. resolve_engine runs in the action worker and receives the trusted ControlPlaneContext plus an opaque saved connection ID; resolve the caller-authorized reference and credentials there. Never put credentials in an action request or return them from the callback.

from etlantic_sql import create_action_handlers
from etlantic_sqlmodel import SQLModelBackendConfig, create_managed_backend

def resolve_engine(ctx, connection_id, *, timeout_seconds):
    # Apply the timeout to both secret lookup and the returned engine's driver
    # connection setup (for psycopg, connect_args={"connect_timeout": ...}).
    return connection_store.resolve_engine(
        ctx, connection_id, timeout_seconds=timeout_seconds
    )

backend = create_managed_backend(
    SQLModelBackendConfig(
        database_url=control_plane_url,
        store_id="control-plane",
        action_handlers=create_action_handlers(resolve_engine),
    ),
    authorizer=authorizer,
)
action_worker = backend.create_action_execution_host(worker_id="actions-1")

Authorize connector.test and connector.schema.inspect for the caller. The connection probe executes only SELECT 1; schema inspection reads at most the requested number of column metadata records (1–100) from the current PostgreSQL schema. Neither handler reads source rows or mutates database objects. The standard managed backend also supplies a revision-pinned, read-only connector.preflight handler when the application has not configured one. It validates and plans the accepted saved definition; it does not promise that a later provider connection or write will succeed. The FastAPI adapter uses the same action handlers and worker API through its create_managed_backend configuration. The connection and schema handlers pass the remaining accepted job deadline as timeout_seconds and require the resolver to accept that keyword. Use it to bound secret resolution and the SQLAlchemy engine's driver-level connection timeout. Existing two-argument resolvers remain compatible with catalog and provisioning actions; update them before enabling connection tests or schema inspection, since cancellation cannot interrupt an in-progress resolver call or DBAPI connection setup before a driver connection exists.

Explicit action provisioning

For deployments that grant connector.provision, the package also provides etlantic_sql.create_action_handlers(resolve_engine). The callback receives the action worker's trusted ControlPlaneContext and an opaque saved connection ID, then returns an application-owned SQLAlchemy engine. Its PostgreSQL table create and cleanup handlers are available alongside the connection-test and schema-inspection handlers described above; connection URLs and credentials stay inside the deployment callback.

Provisioning is create-only. The handler accepts only a typed table schema with safe identifiers, refuses an existing unmanaged table, and records the action ID, owner scope, schema fingerprint and effect ID in the same database. Retries of one accepted action recover the recorded receipt instead of recreating the table. Cleanup requires a worker-verified committed parent provision receipt and removes only that exact effect; its tombstone makes a cleanup retry idempotent. The action factory performs no database writes when constructed, and the Foundry storage inspection handler remains read-only. The SQLite-backed unit case and isolated PostgreSQL loopback qualification exercise create-only conflict, receipt recovery and compensation.

The worker deadline covers the SQL transaction. PostgreSQL actions use local statement and lock timeouts, driver cancellation, and a deadline check before commit. SQLite qualification uses an explicit transaction so rollback also removes DDL. Provider work uses a bounded dedicated daemon thread rather than asyncio's shutdown-blocking default executor. Each handler factory permits one SQL operation at a time; a settling operation cannot create an unbounded queue of database threads. Keep the application-owned engine alive until its pending operation settles.

A confirmed commit retains its verified effect receipt even if acknowledgement arrives after the action deadline. The action remains timed_out, and that receipt can authorize cleanup. If an execution timeout or provider failure has no verified receipt, resubmit the identical action with its original idempotency key and a new deadline. The action keeps its ID, obtains a new worker fence, and recovers its effect registry before create-only provisioning or cleanup. Unknown effects never authorize cleanup without the matching verified receipt.

Entry points: etlantic.source_connectors / sink_connectors / storage_connectors → postgresql.

Examples

python examples/sql_to_sql.py
python examples/sql_boundary_hybrid.py
python examples/sql_transactional_write.py
python examples/sql_failure_recovery.py

Links

SQL tutorial · SQL hello · Source · Issues