kacpermuda commented on code in PR #68588:
URL: https://github.com/apache/airflow/pull/68588#discussion_r3549457958
##########
providers/openlineage/tests/unit/openlineage/utils/test_utils.py:
##########
@@ -36,6 +36,7 @@
from airflow.providers.openlineage.plugins.facets import AirflowDagRunFacet,
AirflowJobFacet
from airflow.providers.openlineage.utils.utils import (
_MAX_DOC_BYTES,
+ _PRODUCER,
Review Comment:
Using the producer here is not needed I think, it will use it by default.
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py:
##########
@@ -126,6 +128,18 @@ def _emit_manual_state_change_event(adapter_method,
stats_key, **kwargs):
return event
+def _build_task_event_job_facet_kwargs(*, task, dag, task_metadata:
OperatorLineage) -> dict:
+ if not get_source_code_location_job_facet(dag):
+ return {}
+ return {
+ "job_facets": build_task_event_job_facets(
+ task=task,
+ dag=dag,
+ additional_job_facets=task_metadata.job_facets,
+ )
+ }
+
+
Review Comment:
Let's try to keep the listener clear from such utils and move them to utils
if needed, or maybe we can get rid of them altogether and use some existing
place to put this logic
##########
airflow-core/docs/core-concepts/dags.rst:
##########
@@ -707,6 +707,32 @@ This is especially useful if your tasks are built
dynamically from configuration
EmptyOperator(task_id="extract_orders")
+Dag Source Code Location
+------------------------
+
+You can annotate a Dag with information about the source repository and
revision that define it.
+This metadata is stored with the serialized Dag and can be used by
integrations that need to link a Dag run back to source code.
+
+.. code-block:: python
+
+ from airflow.sdk import DAG, SourceCodeLocation
+
+ with DAG(
+ "my_dag",
+ schedule="@daily",
+ source_code_location=SourceCodeLocation(
+ repo_url="https://github.com/apache/airflow.git";,
+ path="dags/my_dag.py",
+ version="abc123",
+ branch="main",
+ ),
+ ):
Review Comment:
Any way we can grab all of this information from git bundle for example? It
would remove the need to manually annotate.
##########
providers/openlineage/tests/unit/openlineage/plugins/test_listener.py:
##########
@@ -104,6 +105,13 @@ def direct_submit_call(self, callable, *args, **kwargs):
return callable(*args, **kwargs)
+def assert_source_code_location_job_facet(event):
+ source_code_location_facet = event.job.facets["sourceCodeLocation"]
+ assert source_code_location_facet.url ==
"https://github.com/apache/airflow.git";
+ assert source_code_location_facet.path == "dags/example.py"
+ assert source_code_location_facet.version == "abc123"
+
+
Review Comment:
This should be in the tests itself, no need for asserting in separate
function. Let's keep assertions as close to tests as possible. Also, in some
tests we're keeping in tests and in other we're using this util, so let's be
consistent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
