Update triggering asset events key type

airflow/models/taskinstance.py

@@ -947,7 +947,7 @@ def get_prev_start_date_success() -> pendulum.DateTime | None:
     def get_prev_end_date_success() -> pendulum.DateTime | None:
         return timezone.coerce_datetime(_get_previous_dagrun_success().end_date)
 
-    def get_triggering_events() -> dict[str, list[AssetEvent]]:
+    def get_triggering_events() -> dict[Asset, list[AssetEvent]]:
         if TYPE_CHECKING:
             assert session is not None
 
@@ -958,10 +958,10 @@ def get_triggering_events() -> dict[str, list[AssetEvent]]:
         if dag_run not in session:
             dag_run = session.merge(dag_run, load=False)
         asset_events = dag_run.consumed_asset_events
-        triggering_events: dict[str, list[AssetEvent]] = defaultdict(list)
+        triggering_events: dict[Asset, list[AssetEvent]] = defaultdict(list)
         for event in asset_events:
             if event.asset:
-                triggering_events[event.asset.uri].append(event)
+                triggering_events[event.asset].append(event)
 
         return triggering_events
 

docs/apache-airflow/templates-ref.rst

@@ -79,7 +79,7 @@ Variable                                    Type                  Description
 ``{{ expanded_ti_count }}``                 int | ``None``        | Number of task instances that a mapped task was expanded into. If
                                                                   | the current task is not mapped, this should be ``None``.
                                                                   | Added in version 2.5.
-``{{ triggering_asset_events }}``           dict[str,             | If in an Asset Scheduled DAG, a map of Asset URI to a list of triggering :class:`~airflow.models.asset.AssetEvent`
+``{{ triggering_asset_events }}``           dict[str,             | If in an Asset Scheduled DAG, a map of Asset objects to a list of triggering :class:`~airflow.models.asset.AssetEvent`
                                             list[AssetEvent]]     | (there may be more than one, if there are multiple Assets with different frequencies).
                                                                   | Read more here :doc:`Assets <authoring-and-scheduling/datasets>`.
                                                                   | Added in version 2.4.

task_sdk/src/airflow/sdk/definitions/context.py

@@ -72,7 +72,7 @@ class Context(TypedDict, total=False):
     templates_dict: dict[str, Any] | None
     test_mode: bool
     ti: RuntimeTaskInstanceProtocol
-    # triggering_asset_events: Mapping[str, Collection[AssetEvent | AssetEventPydantic]]
+    # triggering_asset_events: Mapping[Asset, Collection[AssetEvent | AssetEventPydantic]]
     triggering_asset_events: Any
     try_number: int | None
     ts: str