Airflow API/Webserver Management

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

The Airflow API and Webserver management within the Meltano Airflow extension provides a structured way to initialize, configure, and execute Airflow services as a Meltano utility. The extension acts as a wrapper around the standard Airflow CLI, automating the setup of configuration files (airflow.cfg), database migrations, and the deployment of a custom DAG generator that translates Meltano schedules into Airflow DAGs.

This module supports both Airflow 2.x and 3.x, handling specific differences in command syntax and configuration keys. For instance, it manages the transition from AIRFLOW__WEBSERVER__WEB_SERVER_PORT to AIRFLOW__API__PORT and handles the change from db init to db migrate.

Sources: README.md:3-12, airflow_ext/wrapper.py:32-60, airflow_ext/wrapper.py:155-180

Service Architecture and Invocation

The extension uses two primary entry points: airflow_extension for management tasks (like initialize and describe) and airflow_invoker for passing through standard Airflow commands. The Airflow class in wrapper.py serves as the core logic engine, utilizing an Invoker to execute binary commands.

The diagram shows the flow from a Meltano command through the extension's wrapper logic to the final execution of Airflow binaries.
Sources: pyproject.toml:34-36, airflow_ext/wrapper.py:35-40, airflow_ext/pass_through.py:14-22

Configuration Management

The extension dynamically generates airflow.cfg by invoking the default Airflow configuration list. It specifically scrubs potential warning headers (such as deprecation notices) from the output to ensure a valid .ini file is written to the AIRFLOW_CONFIG path.

Key Environment Variables and Settings

Setting Name	Environment Variable	Default Path / Value	Description
Database Conn	AIRFLOW__DATABASE__SQL_ALCHEMY_CONN	sqlite:///$ROOT/.meltano/.../airflow.db	SQLAlchemy connection string
DAGs Folder	AIRFLOW__CORE__DAGS_FOLDER	$ROOT/orchestrate/airflow/dags	Where Airflow looks for DAG files
Webserver Port	AIRFLOW__API__PORT	8080	Port for the API and Webserver (Airflow 3.x)
Airflow Home	AIRFLOW_HOME	$ROOT/orchestrate/airflow	Root directory for Airflow files

Sources: README.md:25-54, airflow_ext/wrapper.py:134-153

Initialization and Database Management

The initialize command performs critical setup tasks to prepare the environment for the webserver and API. It is version-aware, detecting the installed Airflow version to run the correct database setup commands.

Database Initialization Flow

1. Version Detection: Runs airflow version and parses the output.
2. Command Execution:
   • For Airflow 2.x: Executes airflow db init.
   • For Airflow 3.x: Executes airflow db migrate.
3. DAG Generation: Copies the meltano_dag_generator.py (sourced from airflow_ext/files/orchestrate/meltano.py) into the DAGs folder.

Webserver and API Invocation

To start the Airflow web interface and API, the extension provides a direct pass-through mechanism. While the scheduler and triggerer are typically run in the background, the webserver (or api-server in newer versions) is often run in the foreground.

Execution Example

# Start the webserver/api-server
meltano invoke airflow api-server

Sources: README.md:92-100, airflow_ext/main.py:65-79

Automated DAG Orchestration

The management system includes a meltano.py script that is automatically deployed to the Airflow environment. This script queries the Meltano project for schedules and dynamically registers them as Airflow DAGs.

DAG Creation Logic

• Meltano V2 Schedules: Processes both elt tasks and complex job tasks.
• Task Mapping: elt schedules are mapped to a single BashOperator named extract_load. job schedules are mapped to a sequence of BashOperator tasks with upstream dependencies.
• Version Compatibility: Uses schedule (Airflow 3+) or schedule_interval (Airflow 2) dynamically.

Sources: airflow_ext/files/orchestrate/meltano.py:126-150, airflow_ext/files/orchestrate/meltano.py:153-200

Conclusion

Airflow API/Webserver Management in this extension simplifies the operational overhead of running Airflow within a Meltano project. By handling version-specific configuration and database management automatically, it ensures that the webserver and API components are consistently provisioned and configured to interact with the broader Meltano ecosystem.