Fix issue reported by the community in slack with lineage in Cosmos 1.1

cosmos/operators/local.py

@@ -4,6 +4,7 @@
 import shutil
 import signal
 import tempfile
+from attr import define
 from pathlib import Path
 from typing import Any, Callable, Literal, Sequence, TYPE_CHECKING
 
@@ -59,6 +60,13 @@
         logger.exception(error)
         is_openlineage_available = False
 
+        @define
+        class OperatorLineage:  # type: ignore
+            inputs: list[str] = list()
+            outputs: list[str] = list()
+            run_facets: dict[str, str] = dict()
+            job_facets: dict[str, str] = dict()
+
 
 class DbtLocalBaseOperator(DbtBaseOperator):
     """
@@ -309,17 +317,32 @@
         """
         Collect the input, output, job and run facets for this operator.
         It relies on the calculate_openlineage_events_completes having being called before.
+
+        This method is called by Openlineage even if `execute` fails, because `get_openlineage_facets_on_failure`
+        is not implemented.
         """
+
         inputs = []
         outputs = []
         run_facets: dict[str, Any] = {}
         job_facets: dict[str, Any] = {}
 
-        for completed in task_instance.openlineage_events_completes:
-            [inputs.append(input_) for input_ in completed.inputs if input_ not in inputs]  # type: ignore
-            [outputs.append(output) for output in completed.outputs if output not in outputs]  # type: ignore
-            run_facets = {**run_facets, **completed.run.facets}
-            job_facets = {**job_facets, **completed.job.facets}
+        openlineage_events_completes = None
+        if hasattr(self, "openlineage_events_completes"):
+            openlineage_events_completes = self.openlineage_events_completes
+        elif hasattr(task_instance, "openlineage_events_completes"):
+            openlineage_events_completes = task_instance.openlineage_events_completes
+        else:
+            logger.warning("Unable to emit OpenLineage events due to lack of data.")
+
+        if openlineage_events_completes is not None:
+            for completed in openlineage_events_completes:
+                [inputs.append(input_) for input_ in completed.inputs if input_ not in inputs]  # type: ignore
+                [outputs.append(output) for output in completed.outputs if output not in outputs]  # type: ignore
+                run_facets = {**run_facets, **completed.run.facets}
+                job_facets = {**job_facets, **completed.job.facets}
+        else:
+            logger.warning("Unable to emit OpenLineage events due to lack of dependencies or data.")
 
         return OperatorLineage(
             inputs=inputs,

pyproject.toml

@@ -35,6 +35,7 @@ classifiers = [
 ]
 dependencies = [
     # Airflow & Pydantic issue: https://github.com/apache/airflow/issues/32311
+    "attrs",
     "pydantic>=1.10.0,<2.0.0",
     "apache-airflow>=2.3.0",
     "importlib-metadata; python_version < '3.8'",
@@ -133,7 +134,6 @@ dependencies = [
     "apache-airflow-providers-cncf-kubernetes>=5.1.1,<7.3.0",
     "types-PyYAML",
     "types-attrs",
-    "attrs",
     "types-requests",
     "types-python-dateutil",
     "apache-airflow"

tests/operators/test_local.py

@@ -200,6 +200,23 @@ class MockEvent:
     assert facets.job_facets == {"d": 4}
 
 
+def test_run_operator_emits_events_without_openlineage_events_completes(caplog):
+    dbt_base_operator = DbtLocalBaseOperator(
+        profile_config=profile_config,
+        task_id="my-task",
+        project_dir="my/dir",
+        should_store_compiled_sql=False,
+    )
+    delattr(dbt_base_operator, "openlineage_events_completes")
+    facets = dbt_base_operator.get_openlineage_facets_on_complete(dbt_base_operator)
+    assert facets.inputs == []
+    assert facets.outputs == []
+    assert facets.run_facets == {}
+    assert facets.job_facets == {}
+    log = "Unable to emit OpenLineage events due to lack of dependencies or data."
+    assert log in caplog.text
+
+
 def test_store_compiled_sql() -> None:
     dbt_base_operator = DbtLocalBaseOperator(
         profile_config=profile_config,