Skip to content

catalog

etl_craft.services.catalog

The documentation catalog: every pipeline, task, table, rule and script, and how they connect.

build_catalog reads the Engine DB and works out the lineage of every active SQL task (see services.lineage), then joins them into assets:

  • a table for each SQL task's TARGET_OBJECT, each table a SELECT reads, each business rule's TARGET_TABLE, and each ingestion script's TARGET_OBJECT; with its columns, the tasks that write and read it, the rules on it, and its writers' latest run;
  • a task for each active task, with its parameters, documentation, lineage and latest run;
  • a pipeline, a business rule and an ingestion script page each.

Tables are joined by name as lineage names them: lower case, without the active warehouse database. Two kinds of edge connect them: column edges, one per target column and source column, and table edges, one per source table and target table of a task. A SQL task whose columns cannot be traced keeps its table edges, from the tables its SELECT reads, and its reason is listed. An ingestion script appears at table level: a PYTHON task may name the table it writes in TARGET_OBJECT and where it reads from in SOURCE_OBJECT, shown as an external source.

With with_warehouse, each table that exists in the warehouse gets its column types, and comments where the warehouse's information schema has them.

EXTERNAL_PREFIX module-attribute 

EXTERNAL_PREFIX = 'external:'

How the name of a script's external source is told apart from a warehouse table.

Column dataclass

A column of a table: its name, and its type and comment when the warehouse gave them.

TableAsset dataclass

A table, or a script's external source, and everything that touches it.

external property 

external: bool

Whether this is an ingestion script's source rather than a warehouse table.

label property 

label: str

The name shown: an external source without its prefix.

add_column 

add_column(name: str) -> None

Record a column, keeping the order columns are first seen in.

TableEdge dataclass

A task reads source and writes target.

TaskAsset dataclass

An active task: its definition, lineage, dependencies and latest run.

upstream and downstream are the tasks it waits for and that wait for it, each with the dependency type, as PIPELINE_CODE.TASK_CODE; rules are its business rules.

label property 

label: str

PIPELINE_CODE.TASK_CODE.

script property 

script: str | None

The ingestion script a PYTHON task runs.

PipelineAsset dataclass

An active pipeline, its tasks, the pipelines it depends on, and its recent runs.

runs are its latest runs, newest first; run_changes what operators changed in each of them, by run; paused is its open pause.

interventions property 

interventions: list[Intervention]

What operators changed in its last run.

RuleAsset dataclass

An active business rule and the table it checks.

task property 

task: str

The rule's task, PIPELINE_CODE.TASK_CODE.

ScriptAsset dataclass

An ingestion script, the tasks that run it, and whether the file is there.

Catalog dataclass

Every asset, and every edge between tables and between columns.

built_from maps a run to the upstream runs it consumed; consumed_by maps an upstream run to the downstream runs that consumed it.

untraced property 

untraced: list[TaskAsset]

The SQL tasks whose column lineage could not be worked out.

where 

where(table: str) -> tuple[str, str, str]

Return a table's (database, schema, table); the active database when unnamed.

RunKpis dataclass

What a pipeline's or task's recent runs add up to.

success_rate property 

success_rate: float | None

The share of finished runs that succeeded, or None before any finished.

build_catalog 

build_catalog(engine: Engine, config: ConnectorConfig, *, with_warehouse: bool = False) -> Catalog

Read the Engine DB and return the catalog; lineage is stored as lineage stores it.

run_kpis 

run_kpis(runs: Sequence[RunSummary | TaskRunSummary]) -> RunKpis

Return the KPIs of runs: successes, failures, SLA misses, durations and rows.

A run counts as finished once it is SUCCESS, FAILED or CANCELLED; SKIPPED runs did no work and IN-PROGRESS ones have not ended, so neither counts either way.

latest_run 

latest_run(runs: Iterable[LastRun | None]) -> LastRun | None

Return the run that ended last, or None when none has.