Local Execution Patterns

Relevant source files

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

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

Local execution in the Meltano Airflow extension centers on wrapping the Apache Airflow CLI to manage orchestration within a Meltano project environment. This pattern allows developers to initialize Airflow metadata, generate Directed Acyclic Graphs (DAGs) from Meltano schedules, and invoke Airflow components (scheduler, webserver, worker) locally using Meltano-managed configurations and paths.

The system utilizes the Meltano Extension Developer Kit (EDK) to provide two primary entry points: an airflow_extension for lifecycle management (initialize, describe) and an airflow_invoker for direct passthrough to the underlying Airflow binary. This ensures that Airflow operates using the project's specific virtual environment and environment variables, such as AIRFLOW_HOME and AIRFLOW_CONFIG.

Sources: README.md:1-12, airflow_ext/wrapper.py:25-50, pyproject.toml:35-37

Architecture and Command Flow

The local execution pattern is built around the Airflow class, which implements the ExtensionBase interface. The extension manages the transition from Meltano commands to Airflow execution by handling environment setup and configuration generation.

Initialization Pattern

The initialization process (meltano invoke airflow:initialize) is a critical local execution pattern that prepares the environment for Airflow. It performs three primary tasks:

1. Configuration Generation: It runs airflow config list --defaults to capture the default configuration and writes it to the local airflow.cfg file, ensuring compatibility between Airflow 2.x and 3.x.
2. Database Migration: It detects the Airflow version. For version 2.x, it runs db init; for version 3.x, it runs db migrate.
3. DAG Orchestrator Deployment: It copies a bundled meltano.py DAG generator into the project's DAGs folder.

Sources: airflow_ext/wrapper.py:82-113, airflow_ext/wrapper.py:145-191

Execution Environment Variables

Local execution relies on specific environment variables to isolate the Airflow instance within the Meltano project.

Variable	Source / Default	Description
AIRFLOW_HOME	$MELTANO_PROJECT_ROOT/orchestrate/airflow	The root directory for Airflow files.
AIRFLOW_CONFIG	$AIRFLOW_HOME/airflow.cfg	Path to the Airflow configuration file.
AIRFLOW__CORE__DAGS_FOLDER	$AIRFLOW_HOME/dags	Location where DAGs are stored and scanned.
SQL_ALCHEMY_CONN	sqlite:///$MELTANO_PROJECT_ROOT/.meltano/.../airflow.db	The connection string for the metadata DB.

Sources: README.md:30-48, airflow_ext/wrapper.py:40-50

DAG Generation Pattern

The local DAG generator (meltano_dag_generator.py) dynamically creates Airflow DAGs based on the meltano schedule list --format=json output. This pattern supports both legacy Meltano "ELT" schedules and modern "Job" schedules.

Generator Logic Flow

Task Execution Details

In local execution, Airflow tasks are typically executed via the BashOperator. These tasks navigate to the MELTANO_PROJECT_ROOT and execute the meltano binary located in the local .meltano/run/bin directory (or fallback to PATH).

• ELT Tasks: Execute meltano schedule run <name>.
• Job Tasks: Execute meltano run <task_args>, with sequential dependencies mapped between tasks in a job.

Sources: airflow_ext/files/orchestrate/meltano.py:38-44, airflow_ext/files/orchestrate/meltano.py:102-106, airflow_ext/files/orchestrate/meltano.py:151-155

Passthrough Execution Pattern

For standard Airflow operations (e.g., creating users, starting the scheduler), the extension uses a passthrough pattern. The airflow_invoker script acts as a shim that initializes the Airflow class and passes all CLI arguments directly to the underlying airflow binary.

# airflow_ext/pass_through.py:13-20
def pass_through_cli() -> None:
    """Pass through CLI entry point."""
    pass_through_logging_config()
    ext = Airflow()
    ext.pass_through_invoker(
        structlog.getLogger("airflow_invoker"),
        *sys.argv[1:] if len(sys.argv) > 1 else [],
    )

Sources: airflow_ext/pass_through.py:13-20, pyproject.toml:37

This pattern allows users to run commands like:

• meltano invoke airflow scheduler
• meltano invoke airflow users create ...
• meltano invoke airflow version

Sources: README.md:120-140

Summary of Local Patterns

Local execution is designed to be self-contained and automated. By wrapping the Airflow CLI, the extension ensures that the orchestration layer remains synchronized with the Meltano project's schedules and environment, while providing a seamless transition between the two tools via the EDK-based wrapper and invoker.

Sources: README.md:1-10, airflow_ext/wrapper.py:27-35