airflow: render templates at start

Signed-off-by: Maciej Obuchowski <obuchowski.maciej@gmail.com>
OpenLineage · Jun 22, 2022 · 29f19bc · 29f19bc
1 parent ab0c7c5
commit 29f19bc
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 6 deletions.
diff --git a/integration/airflow/openlineage/airflow/listener.py b/integration/airflow/openlineage/airflow/listener.py
@@ -89,6 +89,8 @@ def on_task_instance_running(previous_state, task_instance: "TaskInstance", sess
     parent_run_id = str(uuid.uuid3(uuid.NAMESPACE_URL, f'{dag.dag_id}.{dagrun.run_id}'))
 
     def on_running():
+        task_instance.render_templates()
+
         task_metadata = extractor_manager.extract_metadata(dagrun, task)
 
         adapter.start_task(

diff --git a/...ation/airflow/tests/integration/tests/airflow/dags/postgres_orders_popular_day_of_week.py b/...ation/airflow/tests/integration/tests/airflow/dags/postgres_orders_popular_day_of_week.py
@@ -23,23 +23,30 @@
     'email': ['datascience@example.com']
 }
 
+
+def get_sql() -> str:
+    return '''
+    CREATE TABLE IF NOT EXISTS popular_orders_day_of_week (
+      order_day_of_week VARCHAR(64) NOT NULL,
+      order_placed_on   TIMESTAMP NOT NULL,
+      orders_placed     INTEGER NOT NULL
+    );'''
+
 dag = DAG(
     'postgres_orders_popular_day_of_week',
     schedule_interval='@once',
     default_args=default_args,
+    user_defined_macros={
+        "get_sql": get_sql
+    },
     description='Determines the popular day of week orders are placed.'
 )
 
 
 t1 = PostgresOperator(
     task_id='postgres_if_not_exists',
     postgres_conn_id='food_delivery_db',
-    sql='''
-    CREATE TABLE IF NOT EXISTS popular_orders_day_of_week (
-      order_day_of_week VARCHAR(64) NOT NULL,
-      order_placed_on   TIMESTAMP NOT NULL,
-      orders_placed     INTEGER NOT NULL
-    );''',
+    sql="{{ get_sql() }}",
     dag=dag
 )