generate_yml

etl_craft.services.generate_yml ¶

generate-yml: a pipeline's DAG, or the global trigger DAG, as YAML for an orchestrator.

A pipeline's DAG has one task per active task, which runs etl-craft run --task_code, between an __init__ task (--init-only) and a __finalize__ task (--finalize-only) that runs whatever happened. Each task carries one Airflow trigger_rule, worked out from its run condition and its dependencies' types. Only one run of the DAG goes at a time (max_active_runs: 1), as only one run of a pipeline is IN-PROGRESS.

In remote mode the DAG holds every scheduling rule, since the engine checks none (see execution.remote): a dependency on another pipeline's task is a sensor step before the task, and a dependency on another pipeline is a sensor step after __init__ and before the tasks, unless Global_dag is on and the global DAG triggers the pipelines in order. Rules the orchestrator does not support raise RemoteUnsupportedError.

In local mode the engine checks every rule itself when the run or the task starts. Where Airflow has no exact rule (N, HAS_DATA, or dependencies of mixed types) the permissive all_done is written, and the engine records the task SKIPPED if its condition is not met; dependencies on other pipelines are listed for information.

The DAG's settings come from the pipeline's PIPELINE_PARAMETERS, then Orchestration's DAG defaults, then a built-in default.

RUN_DATE_TEMPLATE `module-attribute` ¶

RUN_DATE_TEMPLATE = '{{ data_interval_end | ds }}'

How each remote DAG step passes the run's date: Airflow's end of the data interval, the day a scheduled run fires, so $$run_date means the same as in local mode, and a backfill by the orchestrator runs each day as of its own date.

TRIGGER_RULES `module-attribute` ¶

TRIGGER_RULES: dict[tuple[str, str], str] = {('ALL', 'SUCCESS'): 'all_success', ('ALL', 'FAILURE'): 'all_failed', ('ALL', 'ALWAYS'): 'all_done', ('ALL', 'HAS_DATA'): 'all_success', ('ANY', 'SUCCESS'): 'one_success', ('ANY', 'FAILURE'): 'one_failed', ('ANY', 'ALWAYS'): 'one_done', ('ANY', 'HAS_DATA'): 'one_success'}

The Airflow trigger rule for a run condition and a dependency type; N is all_done.

trigger_rule ¶

trigger_rule(run_condition: str, dependency_types: list[str]) -> str

Return one trigger rule for a task from its run condition and its dependencies' types.

pipeline_dag ¶

pipeline_dag(conn: Connection, config: ConnectorConfig, pipeline_code: str, *, airflow_version: str = '>=2.2.0') -> dict[str, Any]

Return the DAG of pipeline_code as a dict, in the order it is written.

Raises RemoteUnsupportedError in remote mode for rules the orchestrator does not support.

require_airflow_run_templates ¶

require_airflow_run_templates(version_range: str) -> None

Require a declared version floor supporting run identity and data-interval templates.

The range is one numeric release (2.2.0) or comma-separated constraints of numeric releases (>=2.2,<3, ~=2.3, ==2.10.*); anything else is refused.

global_dag ¶

global_dag(conn: Connection, config: ConnectorConfig) -> dict[str, Any]

Return the DAG that triggers each pipeline's DAG in pipeline-dependency order.

It holds every pipeline that depends on another or is depended on. ConfigurationError unless Orchestration's Global_dag is on; in remote mode, RemoteUnsupportedError for a pipeline with rules the orchestrator does not support.

docs_dag ¶

docs_dag(config: ConnectorConfig) -> dict[str, Any]

Return the DAG that writes the catalog site again on Docs_site.Schedule.

ConfigurationError when no schedule is set.

to_yaml ¶

to_yaml(dag: dict[str, Any], mode: Mode = Mode.LOCAL) -> str

Return a DAG as YAML, under the header that says how to read it in mode.