Initialization Process

Relevant source files

The following files were used as context for generating this wiki page:

• airflow_ext/main.py
• airflow_ext/wrapper.py
• airflow_ext/files/orchestrate/README.md
• airflow_ext/files/orchestrate/meltano.py
• README.md
• pyproject.toml

The initialization process for the airflow-ext utility is a multi-step sequence designed to prepare an Apache Airflow environment within a Meltano project. It ensures that the necessary configuration files exist, the metadata database is migrated to the correct version, and the Meltano DAG generator is deployed to the orchestrator's DAGs folder.

This process is primarily triggered by the initialize command provided by the extension's CLI, which wraps the underlying Airflow class logic to handle environment setup and file provisioning.
Sources: airflow_ext/main.py:24-40, README.md:95-97

Core Initialization Workflow

The initialization logic is encapsulated within the Airflow.initialize method. This method follows a specific order of operations: executing pre-invocation tasks (config and DB setup), creating the DAGs directory, and deploying bundled files.
Sources: airflow_ext/wrapper.py:64-96

Execution Steps

1. Pre-Invoke Tasks: The system first ensures the environment is valid by calling pre_invoke. This step generates the airflow.cfg file and initializes the Airflow database.
2. Directory Creation: It creates the directory specified by AIRFLOW__CORE__DAGS_FOLDER (including parent directories) if it does not already exist.
3. DAG Generator Deployment: It writes the meltano_dag_generator.py file to the DAGs folder by reading the bundled orchestrate/meltano.py resource.
4. Documentation Deployment: It writes a README.md to the DAGs folder to provide instructions on custom DAG definition.

The following diagram illustrates the high-level flow of the initialize command:

Configuration and Database Setup

The extension handles the automated creation of Airflow configuration and database initialization to ensure compatibility across different Airflow versions (2.x and 3.x).
Sources: airflow_ext/wrapper.py:127-172

Configuration Generation

The extension generates an initial airflow.cfg by invoking the airflow config list --defaults command. To prevent file corruption, it specifically filters out any leading warning messages or deprecation notices by searching for the first section header (marked by [).
Sources: airflow_ext/wrapper.py:127-150.%0ASources%3A%20%5Bairflow_ext/wrapper.py%3A127-150)

Database Initialization Logic

The database setup adapts based on the detected Airflow version:

• Airflow 2.x: Executes airflow db init.
• Airflow 3.x: Executes airflow db migrate.

Environment Requirements

The initialization process relies on several environment variables to determine file paths and configuration locations. If AIRFLOW_HOME is missing, the extension will fail to initialize.
Sources: airflow_ext/wrapper.py:38-45

Variable	Description	Default / Fallback
AIRFLOW_HOME	Primary directory for Airflow files	Required (no default)
AIRFLOW_CONFIG	Path to the configuration file	$AIRFLOW_HOME/airflow.cfg
AIRFLOW__CORE__DAGS_FOLDER	Destination for DAGs and the generator	$AIRFLOW_HOME/dags
SLUGIFY_USES_TEXT_UNIDECODE	Compatibility setting for Airflow install	Set to yes during init

Sources: airflow_ext/wrapper.py:38-54, README.md:41-88

Component Overview

Component	Responsibility	Source File
Airflow Class	Orchestrates config generation, DB init, and file deployment.	airflow_ext/wrapper.py
meltano_dag_generator.py	Dynamically generates Airflow DAGs from Meltano schedules.	airflow_ext/files/orchestrate/meltano.py
airflow_extension CLI	Entry point for the initialize command.	airflow_ext/main.py
Invoker	Handles the execution of the underlying airflow binary.	meltano.edk.process (referenced in wrapper.py)

Sources: airflow_ext/main.py:24-40, airflow_ext/wrapper.py:28-125, pyproject.toml:34-36

Summary

The initialization process transforms a standard Meltano utility installation into a functional Airflow environment. By automating the creation of airflow.cfg, handling version-specific database migrations, and deploying a specialized meltano_dag_generator.py, the extension allows users to immediately begin orchestrating Meltano ELT tasks and Jobs through Airflow without manual configuration of the Airflow internals.
Sources: README.md:95-120, airflow_ext/wrapper.py:75-96