feat: Add dynamically emitted asset events to OpenLineage

providers/openlineage/src/airflow/providers/openlineage/extractors/manager.py

@@ -30,6 +30,7 @@
 from airflow.providers.openlineage.extractors.bash import BashExtractor
 from airflow.providers.openlineage.extractors.python import PythonExtractor
 from airflow.providers.openlineage.utils.utils import (
+    get_runtime_outlet_assets,
     get_unknown_source_attribute_run_facet,
     try_import_from_string,
 )
@@ -131,7 +132,7 @@ def extract_metadata(
                     if hook_lineage is not None:
                         task_metadata = task_metadata.merge(hook_lineage)
                     else:  # Last resort - check manual annotations
-                        self.extract_inlets_and_outlets(task_metadata, task)
+                        self.extract_inlets_and_outlets(task_metadata, task, task_instance)
                 return task_metadata
 
             except Exception as e:
@@ -151,7 +152,7 @@ def extract_metadata(
             task_metadata = OperatorLineage(
                 run_facets=get_unknown_source_attribute_run_facet(task=task),
             )
-            self.extract_inlets_and_outlets(task_metadata, task)
+            self.extract_inlets_and_outlets(task_metadata, task, task_instance)
             return task_metadata
 
         return OperatorLineage()
@@ -174,18 +175,30 @@ def _get_extractor(self, task: BaseOperator) -> BaseExtractor | None:
             return extractor(task)
         return None
 
-    def extract_inlets_and_outlets(self, task_metadata: OperatorLineage, task) -> None:
+    def extract_inlets_and_outlets(
+        self,
+        task_metadata: OperatorLineage,
+        task,
+        task_instance=None,
+    ) -> None:
         if task.inlets or task.outlets:
             self.log.debug("Manually extracting lineage metadata from inlets and outlets")
         for i in task.inlets:
-            d = self.convert_to_ol_dataset(i)
-            if d:
+            if d := self.convert_to_ol_dataset(i):
                 task_metadata.inputs.append(d)
         for o in task.outlets:
-            d = self.convert_to_ol_dataset(o)
-            if d:
+            if d := self.convert_to_ol_dataset(o):
                 task_metadata.outputs.append(d)
 
+        # Add runtime-emitted outlets (alias resolutions + dynamic asset events), deduped
+        # by (namespace, name) against both static outputs and each other.
+        seen = {(d.namespace, d.name) for d in task_metadata.outputs}
+        for asset, _alias in get_runtime_outlet_assets(task_instance):
+            ol = translate_airflow_asset(asset, None)
+            if ol is not None and (ol.namespace, ol.name) not in seen:
+                task_metadata.outputs.append(ol)
+                seen.add((ol.namespace, ol.name))
+
     def get_hook_lineage(
         self,
         task_instance=None,
@@ -333,10 +346,13 @@ def convert_to_ol_dataset_from_table(table: Table) -> Dataset:
     def convert_to_ol_dataset(obj) -> Dataset | None:
         from openlineage.client.event_v2 import Dataset
 
+        from airflow.providers.common.compat.assets import Asset
         from airflow.providers.common.compat.lineage.entities import File, Table
 
         if isinstance(obj, Dataset):
             return obj
+        if isinstance(obj, Asset):
+            return translate_airflow_asset(obj, None)
         if isinstance(obj, Table):
             return ExtractorManager.convert_to_ol_dataset_from_table(obj)
         if isinstance(obj, File):

providers/openlineage/src/airflow/providers/openlineage/utils/utils.py

@@ -34,7 +34,7 @@
 
 from airflow import __version__ as AIRFLOW_VERSION
 from airflow.models import DagRun, TaskInstance, TaskReschedule
-from airflow.providers.common.compat.assets import Asset
+from airflow.providers.common.compat.assets import Asset, AssetAlias
 from airflow.providers.common.compat.module_loading import import_string
 from airflow.providers.common.compat.sdk import (
     DAG,
@@ -614,6 +614,7 @@ def __init__(self, obj):
         self._cast_fields()
         self._rename_fields()
         self._include_fields()
+        self._extend_fields()
         dict.__init__(
             self,
             **{field: InfoJsonEncodable._cast_basic_types(getattr(self, field)) for field in self._fields},
@@ -630,6 +631,21 @@ def _cast_basic_types(value):
             return str(list(value))
         return value
 
+    def _extend_fields(self) -> None:
+        """
+        Subclass hook to register extra fields not sourced from ``self.obj``.
+
+        Use this primarily for values passed into ``__init__`` (e.g. caller-supplied
+        context) or values computed on the subclass itself. For anything derived from
+        ``self.obj``, prefer the declarative ``casts`` / ``renames`` / ``includes``
+        machinery — this hook is for fields that don't fit that model.
+
+        Runs after ``_cast_fields`` / ``_rename_fields`` / ``_include_fields`` and
+        before ``dict.__init__`` freezes ``self`` into its serialized form, so
+        implementations should both set the attribute on ``self`` and append its
+        name to ``self._fields``.
+        """
+
     def _rename_fields(self):
         for field, renamed in self.renames.items():
             if hasattr(self.obj, field):
@@ -862,10 +878,81 @@ class TaskInstanceInfo(InfoJsonEncodable):
 
 
 class AssetInfo(InfoJsonEncodable):
-    """Defines encoding Airflow Asset object to JSON."""
+    """
+    Defines encoding Airflow Asset object to JSON.
+
+    Output shape: ``{"uri", "extra", "type"[, "source_alias"]}``.
+    ``type`` distinguishes how the asset showed up on the task:
+
+    * ``"asset"`` - statically declared on ``task.outlets`` / ``task.inlets``.
+    * ``"asset_event"`` — emitted dynamically via ``outlet_events[asset].extra``.
+    * ``"asset_event_from_alias"`` — resolved at runtime from an ``AssetAlias``
+      via ``outlet_events[alias].add(asset, ...)``; ``source_alias`` then carries the alias name.
+
+    ``AssetAlias`` are intentionally NOT included - they are just helpers with no URI of their own.
+    """
 
     includes = ["uri", "extra"]
 
+    def __init__(self, obj, asset_type: str = "asset", source_alias: str | None = None):
+        self.type = asset_type
+        self.source_alias = source_alias
+        super().__init__(obj)
+
+    def _extend_fields(self) -> None:
+        if self.type:
+            self._fields.append("type")
+        if self.source_alias:
+            self._fields.append("source_alias")
+
+
+def get_runtime_outlet_assets(task_instance) -> list[tuple[Asset, AssetAlias | None]]:
+    """
+    Return ``(asset, alias)`` pairs for every runtime-attached outlet.
+
+    * ``outlet_events[AssetAlias].add(asset, ...)`` → ``(asset, alias)``.
+    * ``outlet_events[Asset].extra = {...}`` → ``(asset, None)``.
+
+    Empty list on AF2 or when ``task_instance`` is not an AF3 ``RuntimeTaskInstance``
+    (scheduler / API-server path has no usable worker template context).
+    """
+    if not AIRFLOW_V_3_0_PLUS:
+        return []
+    try:
+        from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
+    except ImportError:
+        return []
+    if not isinstance(task_instance, RuntimeTaskInstance):
+        return []
+    if not hasattr(task_instance, "get_template_context"):
+        return []
+    outlet_events = task_instance.get_template_context().get("outlet_events")
+    if outlet_events is None:
+        return []
+    results: list[tuple[Asset, AssetAlias | None]] = []
+    for key in outlet_events:
+        accessor = outlet_events[key]
+        if isinstance(key, AssetAlias):
+            for event in getattr(accessor, "asset_alias_events", []) or []:
+                # ``dest_asset_extra`` was introduced in Airflow 3.1.4 ; on
+                # earlier versions the alias event carries only ``extra``.
+                dest_asset_extra = getattr(event, "dest_asset_extra", None) or {}
+                results.append(
+                    (
+                        Asset(
+                            name=event.dest_asset_key.name,
+                            uri=event.dest_asset_key.uri,
+                            # Per-event extra wins over asset-level extra.
+                            extra={**dest_asset_extra, **(event.extra or {})},
+                        ),
+                        key,
+                    )
+                )
+        elif isinstance(key, Asset):
+            if extra := (getattr(accessor, "extra", None) or {}):
+                results.append((Asset(name=key.name, uri=key.uri, extra=dict(extra)), None))
+    return results
+
 
 class TaskInfo(InfoJsonEncodable):
     """Defines encoding BaseOperator/AbstractOperator object to JSON."""
@@ -943,6 +1030,25 @@ class TaskInfo(InfoJsonEncodable):
         "operator_provider_version": lambda task: get_operator_provider_version(task),
     }
 
+    def __init__(self, obj, runtime_assets: list[tuple[Asset, AssetAlias | None]] | None = None):
+        self._runtime_assets = runtime_assets  # Will not be included in final dict, not in self._fields.
+        super().__init__(obj)
+
+    def _extend_fields(self) -> None:
+        if not self._runtime_assets:
+            return
+        if not hasattr(self, "outlets"):  # Should never happen, just in case
+            self.outlets = []
+            self._fields.append("outlets")
+        for asset, alias in self._runtime_assets:
+            # Append runtime-resolved outlets (alias resolutions + dynamic asset events).
+            # Inlets are not touched - Airflow has no dynamic inlets.
+            self.outlets.append(
+                AssetInfo(asset, asset_type="asset_event")
+                if alias is None
+                else AssetInfo(asset, asset_type="asset_event_from_alias", source_alias=alias.name)
+            )
+
 
 class TaskInfoComplete(TaskInfo):
     """Defines encoding BaseOperator/AbstractOperator object to JSON used when user enables full task info."""
@@ -1031,12 +1137,17 @@ def get_airflow_run_facet(
     task: BaseOperator,
     task_uuid: str,
 ) -> dict[str, AirflowRunFacet]:
+    runtime_assets = get_runtime_outlet_assets(task_instance)
     return {
         "airflow": AirflowRunFacet(
             dag=DagInfo(dag),
             dagRun=DagRunInfo(dag_run),
             taskInstance=TaskInstanceInfo(task_instance),
-            task=TaskInfoComplete(task) if conf.include_full_task_info() else TaskInfo(task),
+            task=(
+                TaskInfoComplete(task, runtime_assets=runtime_assets)
+                if conf.include_full_task_info()
+                else TaskInfo(task, runtime_assets=runtime_assets)
+            ),
             taskUuid=task_uuid,
         )
     }
@@ -1733,6 +1844,33 @@ def should_use_external_connection(hook) -> bool:
     return True
 
 
+def _get_providers_manager_instance():
+    """
+    Return the ProvidersManager instance to use for OpenLineage converter lookup.
+
+    Picks the task-runtime-safe entry point on Airflow 3.2+ and falls back to the legacy
+    ``airflow.providers_manager.ProvidersManager`` on 3.0 / 3.1 and 2.x. Returns ``None``
+    if neither import is available (very old Airflow).
+
+    Kept as a module-level function so tests can patch this single seam with a plain
+    ``mock.patch(..., return_value=fake_pm)`` instead of juggling ``sys.modules`` or
+    dotted paths that may not exist on the target Airflow version.
+    """
+    try:
+        # Task-runtime-safe entry point — only available in Airflow 3.2+.
+        from airflow.sdk.plugins_manager import ProvidersManagerTaskRuntime
+
+        return ProvidersManagerTaskRuntime()
+    except ImportError:
+        pass
+    try:
+        from airflow.providers_manager import ProvidersManager
+
+        return ProvidersManager()
+    except ImportError:
+        return None
+
+
 def translate_airflow_asset(asset: Asset, lineage_context) -> OpenLineageDataset | None:
     """
     Convert an Asset with an AIP-60 compliant URI to an OpenLineageDataset.
@@ -1746,26 +1884,35 @@ def translate_airflow_asset(asset: Asset, lineage_context) -> OpenLineageDataset
         try:
             from airflow.datasets import _get_normalized_scheme  # type: ignore[no-redef]
         except ImportError:
+            log.debug("Could not import objects required for asset translation.")
             return None
 
-    try:
-        from airflow.providers_manager import ProvidersManager
-
-        ol_converters = getattr(ProvidersManager(), "asset_to_openlineage_converters", None)
-        if not ol_converters:
-            ol_converters = ProvidersManager().dataset_to_openlineage_converters  # type: ignore[attr-defined]
+    providers_manager = _get_providers_manager_instance()
+    if providers_manager is None:
+        log.debug("Could not resolve a ProvidersManager instance for asset translation.")
+        return None
 
-        normalized_uri = asset.normalized_uri
-    except (ImportError, AttributeError):
+    ol_converters = getattr(providers_manager, "asset_to_openlineage_converters", None)
+    if not ol_converters:
+        ol_converters = getattr(providers_manager, "dataset_to_openlineage_converters", None)
+    if not ol_converters:
+        log.debug("Could not retrieve openlineage converters.")
         return None
 
-    if normalized_uri is None:
+    if not (normalized_uri := getattr(asset, "normalized_uri", None)):
+        log.debug("Normalized uri is not accessible or empty.")
         return None
 
     if not (normalized_scheme := _get_normalized_scheme(normalized_uri)):
+        log.debug("Normalized scheme is empty for URI `%s`.", normalized_uri)
         return None
 
     if (airflow_to_ol_converter := ol_converters.get(normalized_scheme)) is None:
+        log.debug("No converter found for scheme `%s`.", normalized_scheme)
         return None
 
-    return airflow_to_ol_converter(Asset(uri=normalized_uri, extra=asset.extra), lineage_context)
+    try:
+        return airflow_to_ol_converter(Asset(uri=normalized_uri, extra=asset.extra), lineage_context)
+    except Exception as err:
+        log.debug("Airflow asset `%s` conversion to OL Dataset failed with err `%s`", normalized_uri, err)
+        return None