How-to guides¶
These guides cover common model construction, rendering, scheduling, resource, and serialization tasks.
How to build a DAG in Python¶
Construct operator-specific task models and place them in a Dag:
from airflow_pydantic import BashTask, Dag, PythonTask
dag = Dag(
dag_id="daily-etl",
schedule="0 2 * * *",
start_date="2025-01-01",
catchup=False,
default_args={"owner": "data", "retries": 2},
tasks={
"extract": BashTask(bash_command="python /opt/etl/extract.py"),
"load": PythonTask(
python_callable="builtins.print",
op_kwargs={"table": "events"},
dependencies=["extract"],
),
},
)
Task dictionary keys become task_id values when the DAG is rendered or
instantiated. dependencies contains upstream task IDs.
How to render deployable Python¶
Render one DAG model and write the result:
from pathlib import Path
source = dag.render()
Path("generated/daily_etl.py").parent.mkdir(parents=True, exist_ok=True)
Path("generated/daily_etl.py").write_text(source)
To render a complete airflow-config collection, use
Configuration.generate() instead. It avoids rewriting unchanged files.
How to instantiate models in memory¶
Instantiate a model inside an Airflow DAG context:
from airflow import DAG
with DAG(dag_id="daily-etl", schedule=None) as airflow_dag:
dag.instantiate(dag=airflow_dag)
For declarative configurations, Configuration.generate_in_mem() registers
every configured DAG in the calling module.
How to save SSH connection options on a host¶
Store shared SSHHook options on the host, then pass the host to an SSH task:
from airflow_pydantic import Host, SSHTask
host = Host(
name="worker.example.com",
username="airflow",
ssh_hook_args={"port": 2222, "conn_timeout": 20, "keepalive_interval": 15},
)
task = SSHTask(task_id="check", command="hostname", ssh_hook=host)
Use the same field in a Hydra host definition:
_target_: airflow_pydantic.Host
name: worker.example.com
username: airflow
ssh_hook_args:
port: 2222
conn_timeout: 20
keepalive_interval: 15
For a one-off hook, override individual options at the call site:
hook = host.hook(conn_timeout=60)
How to label mapped SSH tasks by hostname¶
Use a host filter to create one mapped SSH task instance per matching host:
from airflow_pydantic import BalancerConfiguration, BalancerHostQueryConfiguration, Host, SSHTask
balancer = BalancerConfiguration(
hosts=[
Host(name="worker-a.example.com", username="airflow", tags=["worker"]),
Host(name="worker-b.example.com", username="airflow", tags=["worker"]),
],
)
task = SSHTask(
task_id="check-workers",
command="hostname",
ssh_hook=BalancerHostQueryConfiguration(kind="filter", balancer=balancer, tag="worker"),
)
On Airflow 2.9+ and Airflow 3, the task automatically sets
map_index_template="{{ task.remote_host }}". Instantiated DAGs and generated
Python use the same default. The labels above become worker-a.example.com and
worker-b.example.com. Airflow renders named indexes after each mapped task
executes, so numeric indexes can appear before execution. See
Airflow’s named mapping documentation.
An existing ssh-template-all-hosts template using a host filter gets this
default automatically. Override it on the template or on an individual task:
_target_: airflow_pydantic.SSHTask
task_id: check-workers
template: ${templates.task.ssh-template-all-hosts}
command: hostname
map_index_template: "{{ task.remote_host | upper }}"
Set map_index_template: null to keep numeric indexes. Single-host tasks retain
their existing default. When constructing models under Airflow 2.8, automatic
hostname labels are omitted because that version does not support named mapping.
Source generated without an installed Airflow package uses the named-mapping
default and requires Airflow 2.9+ when loaded.
How to use sensors and branching¶
Mix sensors and operators in one task mapping:
from airflow_pydantic import BranchPythonTask, FileSensor
dag.tasks.update(
{
"wait-for-input": FileSensor(
filepath="/data/incoming/events.json",
poke_interval=30,
timeout=3600,
),
"choose-path": BranchPythonTask(
python_callable="builtins.print",
dependencies=["wait-for-input"],
),
}
)
Refer to the API reference for every operator and sensor model.
How to use custom timetables¶
Pass a timetable model as the DAG schedule:
from datetime import timedelta
from airflow_pydantic import CronTriggerTimetable, MultipleCronTriggerTimetable
dag.schedule = CronTriggerTimetable(
cron="0 9 * * mon-fri",
timezone="America/New_York",
interval=timedelta(hours=1),
)
multiple = MultipleCronTriggerTimetable(
crons=["0 9 * * mon", "0 9 * * thu"],
timezone="America/New_York",
)
Available timetable models also include data-interval, delta, and event timetables.
How to configure pools and variables¶
Use a pool model anywhere a task accepts pool:
from airflow_pydantic import BashTask, Pool, Variable
task = BashTask(
bash_command="python /opt/jobs/large.py",
pool=Pool(pool="warehouse", slots=8, description="Warehouse capacity"),
pool_slots=2,
)
variable = Variable(key="report_settings", deserialize_json=True)
Models defer Airflow database access until runtime. A Variable can therefore
remain serializable inside configuration.
How to define typed DAG parameters¶
Attach Airflow Param definitions to the DAG model:
from airflow_pydantic import Dag, Param
parameterized = Dag(
dag_id="parameterized-report",
params={
"limit": Param(
100,
type="integer",
minimum=1,
title="Row limit",
)
},
)
Tasks can access the value through Airflow templates such as
{{ params.limit }}.
How to run remote and downstream workflows¶
Use the SSH and trigger models for provider-backed operations:
from airflow_pydantic import SSHTask, TriggerDagRunTask
remote = SSHTask(
ssh_conn_id="reports-host",
command="python /opt/jobs/report.py",
)
trigger = TriggerDagRunTask(
trigger_dag_id="downstream-report",
wait_for_completion=True,
dependencies=["remote"],
)
Install the matching Airflow providers in the scheduler and worker environment.
How to reuse task settings in Python¶
Pass an argument model through template:
from airflow_pydantic import BashTask, BashTaskArgs
shell_defaults = BashTaskArgs(
retries=2,
env={"MODE": "production"},
)
report = BashTask(
template=shell_defaults,
bash_command="python /opt/jobs/report.py",
)
Fields on the concrete task override fields inherited from the template.
How to compose shell environments¶
Build activation and working-directory commands with BashCommands:
from airflow_pydantic import BashCommands
command = BashCommands(
commands=["source /opt/venv/bin/activate", "python report.py"],
cwd="/opt/reports",
login=True,
)
print(str(command))
Use in_virtualenv(), in_conda(), in_bash(), and link() for the matching
single-purpose command transformations.
How to serialize and validate models¶
Use normal Pydantic methods for storage and round trips:
payload = dag.model_dump(mode="json")
restored = Dag.model_validate(payload)
json_payload = dag.model_dump_json(indent=2)
restored_from_json = Dag.model_validate_json(json_payload)
Serialization retains the import paths needed to reconstruct operator, callable, timetable, and connection types.
How to use models with airflow-config¶
Select exported model classes with Hydra _target_ entries:
dags:
remote-report:
schedule: "@daily"
tasks:
report:
_target_: airflow_pydantic.SSHTask
ssh_conn_id: reports-host
command: python /opt/jobs/report.py
See the airflow-config tutorial for loading and materialization.
How to integrate long-running and imported schedules¶
Use task models from the runtime integration that owns the job:
airflow-supervisor for supervisord processes modeled by supervisor-pydantic.
airflow-systemd for services modeled by systemd-pydantic.
airflow-cron for schedules modeled by cron-pydantic.
How to attach task callbacks¶
Set callbacks on task models or in TaskArgs used for DAG default_args:
from airflow_pydantic import BashTask
def alert(context):
context["task"].log.error("Task failed: %s", context["exception"])
task = BashTask(
task_id="report",
bash_command="/opt/jobs/report",
on_failure_callback=alert,
)
Use an importable function in your DAG directory for rendered DAGs and YAML configurations. Supply its dotted path, or a list of paths when several callbacks should run:
task_id: report
operator: airflow_pydantic.BashTask
bash_command: /opt/jobs/report
on_failure_callback:
- alerts.report_failed
on_retry_callback: alerts.report_retrying
Callbacks receive Airflow’s task context. Airflow invokes failure callbacks after task execution fails, subject to retries; changing a task’s state through the UI does not invoke them. Check callback errors in the scheduler or task runner logs for your Airflow version.