feat: add fileloc to DAG info in AirflowRunFacet (apache#41311)

Signed-off-by: Kacper Muda <[email protected]>

airflow/providers/openlineage/utils/utils.py

@@ -244,7 +244,7 @@ def _include_fields(self):
 class DagInfo(InfoJsonEncodable):
     """Defines encoding DAG object to JSON."""
 
-    includes = ["dag_id", "description", "owner", "schedule_interval", "start_date", "tags"]
+    includes = ["dag_id", "description", "fileloc", "owner", "schedule_interval", "start_date", "tags"]
     casts = {"timetable": lambda dag: dag.timetable.serialize() if getattr(dag, "timetable", None) else None}
     renames = {"_dag_id": "dag_id"}
 

tests/providers/openlineage/plugins/test_adapter.py

@@ -587,6 +587,7 @@ def test_emit_dag_started_event(mock_stats_incr, mock_stats_timer, generate_stat
                                 "schedule_interval": "86400.0 seconds",
                                 "start_date": "2024-06-01T00:00:00+00:00",
                                 "tags": [],
+                                "fileloc": pathlib.Path(__file__).resolve().as_posix(),
                             },
                             dagRun={
                                 "conf": {},

tests/providers/openlineage/utils/test_utils.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import datetime
+import pathlib
 from unittest.mock import MagicMock, patch
 
 from airflow import DAG
@@ -138,6 +139,7 @@ def test_get_airflow_dag_run_facet():
             dag={
                 "dag_id": "dag",
                 "description": None,
+                "fileloc": pathlib.Path(__file__).resolve().as_posix(),
                 "owner": "airflow",
                 "timetable": {},
                 "schedule_interval": "@once",