pipelines

etl_craft.engine.repository.pipelines ¶

Pipelines: resolving a code, and a pipeline's row.

PIPELINE_PARAMETER_KINDS `module-attribute` ¶

PIPELINE_PARAMETER_KINDS: dict[str, tuple[str, Callable[[object], bool]]] = {'CATCHUP': ('true or false', lambda value: isinstance(value, bool)), 'DEPENDS_ON_PAST': ('true or false', lambda value: isinstance(value, bool)), 'EMAIL_ON_FAILURE': ('true or false', lambda value: isinstance(value, bool)), 'RETRIES': ('a whole number, 0 or more', _is_count), 'RETRY_DELAY_MINUTES': ('a whole number, 0 or more', _is_count), 'TAGS': ('a list of strings', _is_strings), 'EMAIL_RECIPIENTS': ('a list of strings', _is_strings)}

Each PIPELINE_PARAMETERS key, what its value must be, and the test for it.

PipelineDetail `dataclass` ¶

A pipeline's row, with the DAG settings from PIPELINE_PARAMETERS as typed fields.

with_suggestions ¶

with_suggestions(message: str, unknown: str, candidates: list[str]) -> str

Append a "did you mean" list to message when any candidate is close.

resolve_pipeline_id ¶

resolve_pipeline_id(conn: Connection, pipeline_code: str) -> int

Return the id of active pipeline pipeline_code; MetadataError if there is none.

fetch_pipeline_detail ¶

fetch_pipeline_detail(conn: Connection, pipeline_id: int) -> PipelineDetail

Return the row of pipeline_id, which must exist.

PIPELINE_PARAMETERS is JSON: PostgreSQL returns it decoded, SQLite as text.

pipeline_parameter_problems ¶

pipeline_parameter_problems(stored: object) -> tuple[list[str], list[str]]

Check a pipeline's PIPELINE_PARAMETERS as stored; return (problems, unknown keys).

A value of the wrong type is a problem. A key etl-craft does not read is returned apart: it may be a typo, and it is ignored either way.

fetch_pipeline_handlers ¶

fetch_pipeline_handlers(conn: Connection, pipeline_id: int) -> set[str]

Return the handler of every active task in pipeline_id: what a run will connect to.