Airflow Migration

[pingsutw]

Summary

• Introduces the Airflow migration plugin (flyteplugins-airflow) that monkey-patches airflow.DAG and BaseOperator to transparently convert standard Airflow DAG definitions into runnable Flyte tasks
• Extracts _AirflowTaskMixin to DRY shared behavior (dependency arrows, ExecutorSafeguard workaround) between AirflowShellTask and AirflowPythonFunctionTask
• Breaks FlyteDAG.build() into smaller focused methods (_build_downstream_map(), _find_caller_module(), _create_dag_entry())
• Cleans up unused imports, dead code, and replaces global statements with a module-level _state dict

Test plan

• Verified import flyteplugins.airflow.task triggers patches correctly

from pathlib import Path

import flyteplugins.airflow.task  # noqa: F401 — triggers DAG + operator monkey-patches
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

import flyte


def hello_python():
    print("Hello from PythonOperator!")


# Standard Airflow DAG definition — no Flyte-specific changes needed inside the block.
# Pass flyte_env so the generated workflow task uses the right container image.
with DAG(
    dag_id="simple_airflow_workflow",
) as dag:
    t1 = BashOperator(
        task_id="say_hello",
        bash_command='echo "Hello Airflow!"',
    )
    t2 = BashOperator(
        task_id="say_goodbye",
        bash_command='echo "Goodbye Airflow!"',
    )
    t3 = PythonOperator(
        task_id="hello_python",
        python_callable=hello_python,
    )
    t1 >> t2  # t2 runs after t1


if __name__ == "__main__":
    flyte.init_from_config(root_dir=Path(__file__).parent.parent.parent)
    run = flyte.with_runcontext(mode="remote", log_level="10").run(dag)
    print(run.url)