#726 Rewrite airflow plugin for clarity and maintainability Pull request opened by 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 •

pytest tests/flyte/test_airflow_dag.py

— 2/3 pass (1 pre-existing failure in

test_operator_tasks_registered_in_env

) • 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) flyteorg/flyte-sdk