kacpermuda commented on code in PR #75303:
URL: https://github.com/apache/airflow/pull/75303#discussion_r4377689552
##########
providers/standard/src/airflow/providers/standard/utils/openlineage.py:
##########
@@ -80,9 +80,19 @@ def _get_openlineage_parent_info(ti: TaskInstance |
RuntimeTI) -> dict[str, str]
"rootParentJobNamespace": lineage_root_job_namespace(ti),
}
+ try:
+ from airflow.providers.openlineage.plugins.macros import
lineage_job_type, lineage_root_job_type
+ except ImportError: # Job type macros are missing in older versions of
the OpenLineage provider
+ return parent_info
+
+ parent_info["parentJobType"] = lineage_job_type()
+ if (root_job_type := lineage_root_job_type(ti)) is not None:
+ parent_info["rootParentJobType"] = root_job_type
+ return parent_info
+
def _inject_openlineage_parent_info_to_dagrun_conf(
- dr_conf: dict | None, ol_parent_info: dict[str, str]
+ dr_conf: dict | None, ol_parent_info: dict[str, Any]
Review Comment:
Good catch, thanks. Added parentJobType / rootParentJobType to the forbidden
keys and fixed the rootParentJobName / rootParentJobNamespace typo; tests are
parametrized over all keys.
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py:
##########
@@ -55,6 +55,20 @@ def lineage_job_name(task_instance: TaskInstance):
return get_job_name(task_instance)
+def lineage_job_type() -> dict[str, str]:
Review Comment:
Fair point. Documented in macros.rst that these return a dict (or None),
with a | tojson example for string contexts and a note on
render_template_as_native_obj/calling from Python. I didn’t add _json variants,
to keep the API small.
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py:
##########
@@ -154,6 +189,10 @@ def _get_ol_root_id(id_key: str, task_instance:
TaskInstance) -> str | None:
return None
+def _build_airflow_job_type(job_type: Literal["DAG", "TASK"]) -> dict[str,
str]:
Review Comment:
Done. To avoid a circular import, the _JOB_TYPE_DAG / _JOB_TYPE_TASK
constants and a small get_airflow_job_type helper now live in utils.py, and
adapter.py and the macros import them from there.
--
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]
