DagBag loads and parses Airflow DAG definitions from Python files. During parsing, it catches and logs exceptions, allowing tests to inspect import failures and files that contain no DAGs. section 10.1.1
Use DagBag in a DAG integrity test to validate a directory of workflow files:
- Find Python files recursively with glob. section 10.1.1
- Instantiate DagBag with dag_folder set to each file and include_examples=False. section 10.1.1
- Inspect captured logs for parsing errors and reject files reported as containing no DAGs. section 10.1.1
Parameterized integrity check
A pytest test can run once per discovered file: DagBag(dag_folder=dag_file, include_examples=False). If a record has level ERROR, raise its original exception; if the log says no DAGs were found, fail the test. section 10.1.1
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}"
Shows the complete recursive file-discovery and DagBag integrity-test pattern. section 10.1.1
