In an Airflow data pipeline, each task needs a stable primary identifier so Airflow can refer to it consistently across the workflow. The task_id appears in the UI and supports logging, dependency management, and API calls. section 2.2.2
Use task_id to:
- Identify operators such as PythonOperator and BashOperator. section 2.2.2
- Keep generated tasks distinct; assigning one fixed ID inside a loop creates duplicate IDs. section 10.1.1
- Validate DAG integrity before deployment, alongside checks for cycles and other loading errors. section 10.1.1
Naming a Python task
A Python task can use `task_id="get_pictures"` while its callable is `_get_pictures`. Keeping the variable, task ID, and function names similar makes the task easier to recognize in the DAG and Airflow UI. section 2.2.2
Listing 10.3 DAG integrity test
import glob import os import pytest from airflow.models import DagBag DAG_PATH = os.path.join( os.path.dirname(__file__), "..", "..", "dags/**/*.py" ) DAG_FILES = glob.glob(DAG_PATH, recursive=True) @pytest.mark.parametrize("dag_file", DAG_FILES) def test_dag_integrity(dag_file, caplog): """Test integrity of DAGs.""" DagBag(dag_folder=dag_file, include_examples=False) # The DagBag's function catches and logs all exceptions, so we ➥ parse the logs to raise them here for record in caplog.records: if record.levelname == "ERROR": raise record.exc_info[1] elif "assumed to contain no DAGs" in record.message: assert False, f"No DAGs found in {dag_file}"
Inspect a parameterized pytest integrity test that loads every DAG and surfaces errors before deployment, including task-ID uniqueness failures. section 10.1.1
