This is an automated email from the ASF dual-hosted git repository.
mobuchowski pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 6a134fa4d03 dbt Cloud: emit OpenLineage events via the OpenLineage
public API (#74754)
6a134fa4d03 is described below
commit 6a134fa4d03586f68c5b19874a9c1bffe721c076
Author: Kacper Muda <[email protected]>
AuthorDate: Thu Oct 15 10:58:19 2026 +0200
dbt Cloud: emit OpenLineage events via the OpenLineage public API (#74754)
DbtCloudRunJobOperator and DbtCloudJobRunSensor emitted dbt artifact
events by reaching into the OpenLineage listener's adapter directly.
Use the provider's public `emit` API instead, and bump the minimum
required OpenLineage provider version to 2.16.0, where it was added.
Co-authored-by: Claude <[email protected]>
---
providers/dbt/cloud/docs/index.rst | 8 +--
providers/dbt/cloud/pyproject.toml | 2 +-
.../providers/dbt/cloud/utils/openlineage.py | 57 ++++------------------
.../tests/unit/dbt/cloud/utils/test_openlineage.py | 20 ++++----
4 files changed, 24 insertions(+), 63 deletions(-)
diff --git a/providers/dbt/cloud/docs/index.rst
b/providers/dbt/cloud/docs/index.rst
index 10e6491fb28..e5dfe854782 100644
--- a/providers/dbt/cloud/docs/index.rst
+++ b/providers/dbt/cloud/docs/index.rst
@@ -143,11 +143,11 @@ Install them when installing from PyPI. For example:
pip install apache-airflow-providers-dbt-cloud[openlineage]
-=============== ===============================================
+=============== ================================================
Extra Dependencies
-=============== ===============================================
-``openlineage`` ``apache-airflow-providers-openlineage>=2.3.0``
-=============== ===============================================
+=============== ================================================
+``openlineage`` ``apache-airflow-providers-openlineage>=2.16.0``
+=============== ================================================
Downloading official packages
-----------------------------
diff --git a/providers/dbt/cloud/pyproject.toml
b/providers/dbt/cloud/pyproject.toml
index 95be8c0a2f5..e709ad1b2c9 100644
--- a/providers/dbt/cloud/pyproject.toml
+++ b/providers/dbt/cloud/pyproject.toml
@@ -72,7 +72,7 @@ dependencies = [
[project.optional-dependencies]
# pip install apache-airflow-providers-dbt-cloud[openlineage]
"openlineage" = [
- "apache-airflow-providers-openlineage>=2.3.0",
+ "apache-airflow-providers-openlineage>=2.16.0",
]
[dependency-groups]
diff --git
a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/utils/openlineage.py
b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/utils/openlineage.py
index d9336d07189..8f4ff7a7797 100644
--- a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/utils/openlineage.py
+++ b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/utils/openlineage.py
@@ -24,7 +24,6 @@ from typing import TYPE_CHECKING
from airflow.providers.common.compat.openlineage.check import
require_openlineage_version
from airflow.providers.dbt.cloud.hooks.dbt import DBT_CAUSE_MAX_LENGTH
-from airflow.providers.dbt.cloud.version_compat import AIRFLOW_V_3_0_PLUS
if TYPE_CHECKING:
from airflow.models.taskinstance import TaskInstance
@@ -36,28 +35,6 @@ if TYPE_CHECKING:
log = logging.getLogger(__name__)
-def _get_logical_date(task_instance):
- # todo: remove when min airflow version >= 3.0
- if AIRFLOW_V_3_0_PLUS:
- dagrun = task_instance.get_template_context()["dag_run"]
- return dagrun.logical_date or dagrun.run_after
-
- if hasattr(task_instance, "logical_date"):
- date = task_instance.logical_date
- else:
- date = task_instance.execution_date
-
- return date
-
-
-def _get_dag_run_clear_number(task_instance):
- # todo: remove when min airflow version >= 3.0
- if AIRFLOW_V_3_0_PLUS:
- dagrun = task_instance.get_template_context()["dag_run"]
- return dagrun.clear_number
- return task_instance.dag_run.clear_number
-
-
def _get_parent_run_metadata(task_instance):
"""
Retrieve the ParentRunMetadata associated with a specific Airflow task
instance.
@@ -71,31 +48,18 @@ def _get_parent_run_metadata(task_instance):
lineage_job_name,
lineage_job_namespace,
lineage_root_job_name,
+ lineage_root_job_namespace,
lineage_root_run_id,
lineage_run_id,
)
- parent_run_id = lineage_run_id(task_instance)
- parent_job_name = lineage_job_name(task_instance)
- parent_job_namespace = lineage_job_namespace()
-
- root_parent_run_id = lineage_root_run_id(task_instance)
- rot_parent_job_name = lineage_root_job_name(task_instance)
-
- try: # Added in OL provider 2.9.0, try to use it if possible
- from airflow.providers.openlineage.plugins.macros import
lineage_root_job_namespace
-
- root_parent_job_namespace = lineage_root_job_namespace(task_instance)
- except ImportError:
- root_parent_job_namespace = lineage_job_namespace()
-
return ParentRunMetadata(
- run_id=parent_run_id,
- job_name=parent_job_name,
- job_namespace=parent_job_namespace,
- root_parent_run_id=root_parent_run_id,
- root_parent_job_name=rot_parent_job_name,
- root_parent_job_namespace=root_parent_job_namespace,
+ run_id=lineage_run_id(task_instance),
+ job_name=lineage_job_name(task_instance),
+ job_namespace=lineage_job_namespace(),
+ root_parent_run_id=lineage_root_run_id(task_instance),
+ root_parent_job_name=lineage_root_job_name(task_instance),
+ root_parent_job_namespace=lineage_root_job_namespace(task_instance),
)
@@ -155,7 +119,7 @@ def
inject_parent_job_information_into_dbt_cloud_cause(cause: str | None, task_i
return cause
-@require_openlineage_version(provider_min_version="2.5.0")
+@require_openlineage_version(provider_min_version="2.16.0")
def generate_openlineage_events_from_dbt_cloud_run(
operator: DbtCloudRunJobOperator | DbtCloudJobRunSensor, task_instance:
TaskInstance
) -> OperatorLineage:
@@ -175,9 +139,9 @@ def generate_openlineage_events_from_dbt_cloud_run(
"""
from openlineage.common.provider.dbt import DbtCloudArtifactProcessor
+ from airflow.providers.openlineage.api import emit
from airflow.providers.openlineage.extractors import OperatorLineage
from airflow.providers.openlineage.plugins.adapter import _PRODUCER
- from airflow.providers.openlineage.plugins.listener import
get_openlineage_listener
# if no account_id set this will fallback
log.debug("Retrieving information about DBT job run.")
@@ -236,7 +200,6 @@ def generate_openlineage_events_from_dbt_cloud_run(
log.debug("Preparing OpenLineage parent job information to be included in
DBT events.")
parent_metadata = _get_parent_run_metadata(task_instance)
- adapter = get_openlineage_listener().adapter
# process each step in loop, sending generated events in the same order as
steps
for counter, artifacts in enumerate(step_artifacts, 1):
@@ -266,7 +229,7 @@ def generate_openlineage_events_from_dbt_cloud_run(
log.debug("Found %s OpenLineage events for artifact no. %s.",
len(events), counter)
for event in events:
- adapter.emit(event=event)
+ emit(event)
log.debug("Emitted all OpenLineage events for artifact no. %s.",
counter)
log.info("OpenLineage has successfully finished processing information
about DBT job run.")
diff --git a/providers/dbt/cloud/tests/unit/dbt/cloud/utils/test_openlineage.py
b/providers/dbt/cloud/tests/unit/dbt/cloud/utils/test_openlineage.py
index b3f4142dd58..2f38284efba 100644
--- a/providers/dbt/cloud/tests/unit/dbt/cloud/utils/test_openlineage.py
+++ b/providers/dbt/cloud/tests/unit/dbt/cloud/utils/test_openlineage.py
@@ -101,13 +101,14 @@ def get_dbt_artifact(*args, **kwargs):
[
("1.99.0", True),
("2.0.0", True),
- ("2.3.0", True),
- ("2.5.0", False),
+ ("2.5.0", True),
+ ("2.15.0", True),
+ ("2.16.0", False),
("2.99.0", False),
],
)
def test_previous_version_openlineage_provider(value, is_error):
- """When using OpenLineage, the dbt-cloud provider now depends on
openlineage provider >= 2.4"""
+ """When using OpenLineage, the dbt-cloud provider now depends on
openlineage provider >= 2.16"""
def _mock_version(package):
if package == "apache-airflow-providers-openlineage":
@@ -118,7 +119,7 @@ def test_previous_version_openlineage_provider(value,
is_error):
mock_task_instance = MagicMock()
expected_err = (
- f"OpenLineage provider version `{value}` is lower than required
`2.5.0`, "
+ f"OpenLineage provider version `{value}` is lower than required
`2.16.0`, "
"skipping function `generate_openlineage_events_from_dbt_cloud_run`
execution"
)
@@ -160,7 +161,7 @@ def test_get_parent_run_metadata():
class TestGenerateOpenLineageEventsFromDbtCloudRun:
@patch("importlib.metadata.version", return_value="3.0.0")
-
@patch("airflow.providers.openlineage.plugins.listener.get_openlineage_listener")
+ @patch("airflow.providers.openlineage.api.emit")
@patch("airflow.providers.openlineage.plugins.adapter.OpenLineageAdapter.build_task_instance_run_id")
@patch("airflow.providers.openlineage.plugins.adapter.OpenLineageAdapter.build_dag_run_id")
@patch.object(DbtCloudHook, "get_job_run")
@@ -173,7 +174,7 @@ class TestGenerateOpenLineageEventsFromDbtCloudRun:
mock_get_job_run,
mock_build_dag_run_id,
mock_build_task_instance_run_id,
- mock_get_openlineage_listener,
+ mock_emit,
mock_version,
):
mock_operator = MagicMock(spec=DbtCloudRunJobOperator)
@@ -210,15 +211,12 @@ class TestGenerateOpenLineageEventsFromDbtCloudRun:
mock_task_instance.dag_id = DAG_ID
mock_task_instance.dag_run.clear_number = 0
- mock_adapter = MagicMock()
-
- mock_adapter.emit.side_effect = emit_event
- mock_get_openlineage_listener.return_value.adapter = mock_adapter
+ mock_emit.side_effect = emit_event
mock_build_task_instance_run_id.return_value = TASK_UUID
mock_build_dag_run_id.return_value = DAG_UUID
generate_openlineage_events_from_dbt_cloud_run(mock_operator,
task_instance=mock_task_instance)
- assert mock_adapter.emit.call_count == 4
+ assert mock_emit.call_count == 4
def test_do_not_raise_error_if_runid_not_set_on_operator(self):
operator = DbtCloudRunJobOperator(task_id="dbt-job-runid-taskid",
job_id=1500)
