session

etl_craft.handlers.sql.session ¶

One SQL task's work on the warehouse: statements, scratch tables and table shapes.

Every statement goes through Session.run, which logs its step with the row count the driver reports and how long it took, and the SQL itself at DEBUG. A database error becomes a HandlerError naming the action, the target and the step, with the failing SQL written to the attempt's log, so a failed task can be debugged from its log alone.

Counts reported to AUD_TASK_RUN_LOG never come from a driver's rowcount, which some drivers do not report per statement; each is its own COUNT(*), taken before the statement it describes changes what it counts.

ROW_ID_COLUMN `module-attribute` ¶

ROW_ID_COLUMN = 'ROW_ID'

The engine-generated key every table the engine creates gets.

Session ¶

A SQL task's connection to the warehouse, and what it knows about the task's target.

init ¶

__init__(conn: Connection, dialect: WarehouseDialect, *, catalog: str, action: str, target_object: str, task_run_id: int, params: Mapping[str, str]) -> None

Work on target_object (schema.table) in catalog over conn.

run ¶

run(sql: str, params: Mapping[str, Any] | None = None, *, step: str) -> CursorResult[Any]

Run one statement, logging it; HandlerError naming step when it fails.

count ¶

count(sql: str, *, step: str) -> int

Return the single integer sql selects.

create_table_as ¶

create_table_as(name: str, select_sql: str, *, step: str) -> None

CREATE TABLE ... AS SELECT in the task's dialect and table format.

qualify ¶

qualify(object_ref: str) -> str

Return database.schema.table for schema.table, in the target's database.

scratch ¶

scratch(suffix: str, *, persistent: bool = False) -> str

Name a scratch table for this attempt: etl_<suffix>_<task_run_id>_<token>.

token is random per session, so no two attempts, of this task or another, share a name. A temporary table lives in the session's own namespace and stays unqualified; persistent candidates and warehouses without temporary tables use the target's catalog and schema. Every scratch table is recorded, so sweep drops it whatever happens.

create_scratch ¶

create_scratch(name: str, body: str, *, step: str) -> None

Create a scratch table from body, TEMPORARY where the dialect has them.

drop ¶

drop(name: str) -> None

Drop a table if it exists.

sweep ¶

sweep() -> None

Drop every scratch table, ignoring failures.

It runs while an error may already be on its way out; a cleanup failure must not replace the error worth reading. On a warehouse without rollback (Trino) a scratch table left behind would be a real table in the team's schema.

columns ¶

columns(name: str) -> list[tuple[str, str]]

Return [(column, data_type)] of table name in column order.

name is catalog.schema.table, schema.table or a bare temporary scratch table name, matched by name alone: a scratch name carries a token no other table has.

column_types ¶

column_types(name: str) -> dict[str, str]

Read the table's complete types once, with lowercase column keys.

hash_types ¶

hash_types(name: str) -> dict[str, str]

Return types and declared decimal scales for hashing table name.

name is catalog.schema.table, schema.table or a bare temporary scratch table name, matched by name alone: a scratch name carries a token no other table has.

hash ¶

hash(values: list[str], columns: tuple[str, ...]) -> str

Hash values using the target columns' persisted types and decimal scales.

target_columns ¶

target_columns() -> list[tuple[str, str]]

Return the target's columns; empty when it does not exist.

prepare_update_stage ¶

prepare_update_stage(stage: str, keys: tuple[str, ...]) -> None

Prepare a stage for joined updates using its dialect's indexes and statistics.

mutation_target ¶

mutation_target() -> tuple[str, str]

Return (target clause, qualifier) for a correlated UPDATE or DELETE on the target.

Trino rejects an alias on UPDATE and DELETE, so there the table name qualifies instead.

rename ¶

rename(qualified_from: str, qualified_to: str) -> None

Rename a table in the dialect's RENAME syntax.

row_id_insert_parts ¶

row_id_insert_parts() -> tuple[str, str]

Return the extra (columns, values) an INSERT needs where ROW_ID does not fill itself.

An identity column or a sequence default fills ROW_ID. Targets without a generator supply the largest ROW_ID present plus a row number under the target mutation lock. The base is read before the INSERT, since engines disagree about reading the table a statement writes.

check_target_format ¶

check_target_format() -> None

Refuse a format mismatch before creating or evolving an existing target.

row_id_generated ¶

row_id_generated() -> bool

Read the existing target's generator rather than assuming new-table defaults.