This is an automated email from the ASF dual-hosted git repository.
bbovenzi 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 7b163446c17 Link asset events emitted in a task loop to their
iteration (#76840)
7b163446c17 is described below
commit 7b163446c170e42da875f99099924f400af3b77f
Author: ccoliu <[email protected]>
AuthorDate: Thu Nov 12 03:56:20 2026 +0800
Link asset events emitted in a task loop to their iteration (#76840)
* Link asset events emitted in a task loop to their iteration
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* Require hand-built task instance links to pass their region
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../api_fastapi/core_api/datamodels/assets.py | 8 +++
.../core_api/openapi/v2-rest-api-generated.yaml | 11 ++++
.../api_fastapi/core_api/routes/public/assets.py | 4 +-
.../api_fastapi/core_api/routes/public/dag_run.py | 4 +-
.../api_fastapi/core_api/services/public/assets.py | 12 ++++-
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 23 ++++++++
.../airflow/ui/openapi-gen/requests/types.gen.ts | 2 +
.../ui/src/components/Assets/AssetEvent.test.tsx | 33 ++++++++++++
.../ui/src/components/Assets/AssetEvent.tsx | 11 +++-
airflow-core/src/airflow/ui/src/utils/links.ts | 16 ++++--
.../core_api/routes/public/test_assets.py | 61 +++++++++++++++++++++-
.../core_api/routes/public/test_dag_run.py | 2 +
.../src/airflowctl/api/datamodels/generated.py | 2 +
13 files changed, 178 insertions(+), 11 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
index 59d75a66e20..c83f15298b3 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
@@ -216,6 +216,14 @@ class AssetEventResponse(BaseModel):
source_run_id: str | None = None
source_map_index: int
source_task_instance_id: UUID | None = None
+ # The source task instance's coordinates, needed to address it when it ran
inside a task loop or
+ # a mapped expansion. None when the event has no source task instance or
it no longer exists.
+ source_region_id: UUID | None = Field(
+ validation_alias=AliasPath("source_task_instance", "region_id"),
default=None
+ )
+ source_region_index: int | None = Field(
+ validation_alias=AliasPath("source_task_instance", "region_index"),
default=None
+ )
created_dagruns: list[DagRunAssetReference]
timestamp: datetime
partition_key: str | None = None
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index c773fd1c030..194eb2fb1c1 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -12887,6 +12887,17 @@ components:
format: uuid
- type: 'null'
title: Source Task Instance Id
+ source_region_id:
+ anyOf:
+ - type: string
+ format: uuid
+ - type: 'null'
+ title: Source Region Id
+ source_region_index:
+ anyOf:
+ - type: integer
+ - type: 'null'
+ title: Source Region Index
created_dagruns:
items:
$ref: '#/components/schemas/DagRunAssetReference'
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
index 18e02bfc8b1..1a586f8f240 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
@@ -80,7 +80,7 @@ from airflow.api_fastapi.core_api.security import (
requires_access_asset_alias,
requires_access_dag,
)
-from airflow.api_fastapi.core_api.services.public.assets import
serialize_asset_events
+from airflow.api_fastapi.core_api.services.public.assets import
load_source_region, serialize_asset_events
from airflow.api_fastapi.logging.decorators import action_logging
from airflow.assets.manager import asset_manager
from airflow.configuration import conf
@@ -479,7 +479,7 @@ def get_asset_events(
)
assets_event_select = assets_event_select.options(
- subqueryload(AssetEvent.created_dagruns), joinedload(AssetEvent.asset)
+ subqueryload(AssetEvent.created_dagruns),
joinedload(AssetEvent.asset), load_source_region()
)
# Materialize here (not lazily during response serialization) so the
regexp query runs while the
# dependency-applied timeout is still active.
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
index 4ddd7fd0add..604243ef557 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
@@ -110,7 +110,7 @@ from airflow.api_fastapi.core_api.security import (
requires_access_dag_run_bulk,
requires_access_dag_run_clear_bulk,
)
-from airflow.api_fastapi.core_api.services.public.assets import
serialize_asset_events
+from airflow.api_fastapi.core_api.services.public.assets import
load_source_region, serialize_asset_events
from airflow.api_fastapi.core_api.services.public.dag_run import (
BulkDagRunService,
DagRunWaiter,
@@ -307,7 +307,7 @@ def get_upstream_asset_events(
statement=select(AssetEvent)
.join(dagrun_asset_event_table, dagrun_asset_event_table.c.event_id ==
AssetEvent.id)
.where(dagrun_asset_event_table.c.dag_run_id == dag_run_pk)
- .options(joinedload(AssetEvent.asset),
subqueryload(AssetEvent.created_dagruns))
+ .options(joinedload(AssetEvent.asset),
subqueryload(AssetEvent.created_dagruns), load_source_region())
.order_by(AssetEvent.id),
filters=[readable_asset_events_filter,
readable_asset_events_by_asset_filter],
)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/services/public/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/services/public/assets.py
index c7816ebc577..d769ebb809b 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/services/public/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/services/public/assets.py
@@ -19,13 +19,23 @@ from __future__ import annotations
from collections.abc import Sequence
from typing import TYPE_CHECKING
+from sqlalchemy.orm import joinedload
+
from airflow.api_fastapi.common.db.assets import resolve_triggering_event_refs
from airflow.api_fastapi.core_api.datamodels.assets import AssetEventResponse
+from airflow.models.asset import AssetEvent
+from airflow.models.taskinstance import TaskInstance
if TYPE_CHECKING:
from sqlalchemy.orm import Session
+ from sqlalchemy.orm.strategy_options import _AbstractLoad
+
- from airflow.models.asset import AssetEvent
+def load_source_region() -> _AbstractLoad:
+ """Load the source task instance's coordinates that ``AssetEventResponse``
reports, in the same query."""
+ return joinedload(AssetEvent.source_task_instance).load_only(
+ TaskInstance.region_id, TaskInstance.region_index
+ )
def serialize_asset_events(
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 9c98e2d0ba9..1f13f5d7946 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -297,6 +297,29 @@ export const $AssetEventResponse = {
],
title: 'Source Task Instance Id'
},
+ source_region_id: {
+ anyOf: [
+ {
+ type: 'string',
+ format: 'uuid'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Source Region Id'
+ },
+ source_region_index: {
+ anyOf: [
+ {
+ type: 'integer'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Source Region Index'
+ },
created_dagruns: {
items: {
'$ref': '#/components/schemas/DagRunAssetReference'
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index e521da1b7f7..ffed3367ccb 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -79,6 +79,8 @@ export type AssetEventResponse = {
source_run_id?: string | null;
source_map_index: number;
source_task_instance_id?: string | null;
+ source_region_id?: string | null;
+ source_region_index?: number | null;
created_dagruns: Array<DagRunAssetReference>;
timestamp: string;
partition_key?: string | null;
diff --git
a/airflow-core/src/airflow/ui/src/components/Assets/AssetEvent.test.tsx
b/airflow-core/src/airflow/ui/src/components/Assets/AssetEvent.test.tsx
index 1bb7266d84d..582f2120e78 100644
--- a/airflow-core/src/airflow/ui/src/components/Assets/AssetEvent.test.tsx
+++ b/airflow-core/src/airflow/ui/src/components/Assets/AssetEvent.test.tsx
@@ -46,4 +46,37 @@ describe("AssetEvent", () => {
expect(screen.getByText("dagRun.partitionKey: foo")).toBeInTheDocument();
});
+
+ const sourceEvent = {
+ ...baseEvent,
+ source_dag_id: "producer",
+ source_run_id: "run_1",
+ source_task_id: "body.produce",
+ } satisfies Partial<AssetEventResponse> as AssetEventResponse;
+
+ it("links the source to the task instance", () => {
+ render(<AssetEvent assetId={1} event={sourceEvent} />, { wrapper: Wrapper
});
+
+ expect(screen.getByRole("link", { name: "producer" })).toHaveAttribute(
+ "href",
+ "/dags/producer/runs/run_1/tasks/body.produce",
+ );
+ });
+
+ it("links the source to the iteration that emitted the event inside a task
loop", () => {
+ const regionId = "01a1c466-2f84-76be-aac3-2aca74761c14";
+
+ render(
+ <AssetEvent
+ assetId={1}
+ event={{ ...sourceEvent, source_region_id: regionId,
source_region_index: 2 }}
+ />,
+ { wrapper: Wrapper },
+ );
+
+ expect(screen.getByRole("link", { name: "producer" })).toHaveAttribute(
+ "href",
+
`/dags/producer/runs/run_1/tasks/body.produce?region_id=${regionId}®ion_index=2`,
+ );
+ });
});
diff --git a/airflow-core/src/airflow/ui/src/components/Assets/AssetEvent.tsx
b/airflow-core/src/airflow/ui/src/components/Assets/AssetEvent.tsx
index 882d439e5e1..2e1254b8aba 100644
--- a/airflow-core/src/airflow/ui/src/components/Assets/AssetEvent.tsx
+++ b/airflow-core/src/airflow/ui/src/components/Assets/AssetEvent.tsx
@@ -27,6 +27,8 @@ import { Tooltip } from "src/system-components";
import Time from "src/components/Time";
+import { getTaskInstanceLink } from "src/utils/links";
+
import RenderedJsonField from "../RenderedJsonField";
import { TriggeredRuns } from "./TriggeredRuns";
@@ -84,7 +86,14 @@ export const AssetEvent = ({
<Box>{translate("source")}: </Box>
{source === "" ? (
<Link
-
to={`/dags/${event.source_dag_id}/runs/${event.source_run_id}/tasks/${event.source_task_id}${event.source_map_index
> -1 ? `/mapped/${event.source_map_index}` : ""}`}
+ to={getTaskInstanceLink({
+ dag_id: event.source_dag_id ?? "",
+ dag_run_id: event.source_run_id ?? "",
+ map_index: event.source_map_index,
+ region_id: event.source_region_id ?? undefined,
+ region_index: event.source_region_index ?? undefined,
+ task_id: event.source_task_id ?? "",
+ })}
>
<Box color="fg.info" overflowWrap="anywhere" padding={0}>
{event.source_dag_id}
diff --git a/airflow-core/src/airflow/ui/src/utils/links.ts
b/airflow-core/src/airflow/ui/src/utils/links.ts
index 5d99cf9c3f1..fdb5c8c9af1 100644
--- a/airflow-core/src/airflow/ui/src/utils/links.ts
+++ b/airflow-core/src/airflow/ui/src/utils/links.ts
@@ -31,15 +31,23 @@ import { routerConfig, taskInstanceRoutes } from
"src/router";
// Links address the current execution at these coordinates; a try_number
would pin the page to one attempt.
export const getTaskInstanceLink = (
tiOrParams:
- | Pick<
- TaskInstanceResponse,
- "dag_id" | "dag_run_id" | "map_index" | "region_id" | "region_index" |
"task_id"
- >
+ | TaskInstanceResponse
| {
dagId: string;
dagRunId: string;
mapIndex?: number;
taskId: string;
+ }
+ | {
+ readonly dag_id: string;
+ readonly dag_run_id: string;
+ readonly map_index: number;
+ // Required when the task instance is described by hand: one inside a
task loop can only be
+ // addressed with its region, so the caller passes undefined
deliberately when there is none.
+ // A TaskInstanceResponse already carries its region, omitted only
when it has none.
+ readonly region_id: string | undefined;
+ readonly region_index: number | undefined;
+ readonly task_id: string;
},
tab?: TaskInstanceTabValue,
): string => {
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
index 9ae5b2505ed..0d3c362ad0e 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
@@ -50,11 +50,14 @@ from airflow.models.asset import (
from airflow.models.base import ID_LEN
from airflow.models.dagbundle import DagBundleModel
from airflow.models.dagrun import DagRun
+from airflow.models.dynamic_region import DynamicRegion
from airflow.models.serialized_dag import SerializedDagModel
+from airflow.models.taskinstance import TaskInstance
from airflow.models.team import Team
from airflow.models.trigger import Trigger
from airflow.providers.standard.operators.empty import EmptyOperator
-from airflow.sdk import Asset
+from airflow.sdk import Asset, task_group
+from airflow.sdk.definitions._internal.loop import create_loop
from airflow.timetables.simple import PartitionedAtRuntime
from airflow.timetables.trigger import CronPartitionTimetable
from airflow.utils.session import provide_session
@@ -1200,6 +1203,49 @@ class TestGetAssetEvents(TestAssets):
assert response.status_code == 200, response.text
assert [row["source_task_instance_id"] for row in
response.json()["asset_events"]] == [None]
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_reports_the_source_task_instance_region(self, test_client,
dag_maker, session):
+ """An event emitted inside a task loop carries the region the UI needs
to link to its iteration."""
+
+ @task_group
+ def body():
+ EmptyOperator(task_id="produce")
+
+ with dag_maker("loop_producer", serialized=True):
+ loop = create_loop(body, max_iterations=3)
+ run = dag_maker.create_dagrun(run_id="loop_run")
+ region =
session.scalar(select(DynamicRegion).where(DynamicRegion.node_id ==
loop.group_id))
+ source = TaskInstance(
+ task=dag_maker.dag.get_task("body.produce"),
+ run_id=run.run_id,
+ dag_version_id=run.created_dag_version_id,
+ region_id=region.id,
+ region_index=2,
+ )
+ session.add(source)
+ session.flush()
+ (asset,) = _create_assets(session, num=1)
+ session.add(
+ AssetEvent(
+ asset_id=asset.id,
+ source_dag_id=run.dag_id,
+ source_run_id=run.run_id,
+ source_task_id="body.produce",
+ source_map_index=-1,
+ source_task_instance_id=source.id,
+ timestamp=DEFAULT_DATE,
+ )
+ )
+ session.commit()
+
+ response = test_client.get("/assets/events")
+
+ assert response.status_code == 200, response.text
+ (event,) = response.json()["asset_events"]
+ assert event["source_task_instance_id"] == str(source.id)
+ assert event["source_region_id"] == str(region.id)
+ assert event["source_region_index"] == 2
+
def test_should_respond_200(self, test_client, session):
asset1, asset2 = self.create_assets(session=session)
self.create_assets_events(session=session)
@@ -1230,6 +1276,8 @@ class TestGetAssetEvents(TestAssets):
"source_run_id": "source_run_id_1",
"source_map_index": -1,
"source_task_instance_id": None,
+ "source_region_id": None,
+ "source_region_index": None,
"created_dagruns": [
{
"run_id": "source_run_id_1",
@@ -1259,6 +1307,8 @@ class TestGetAssetEvents(TestAssets):
"source_run_id": "source_run_id_2",
"source_map_index": -1,
"source_task_instance_id": None,
+ "source_region_id": None,
+ "source_region_index": None,
"created_dagruns": [
{
"run_id": "source_run_id_2",
@@ -1479,6 +1529,8 @@ class TestGetAssetEvents(TestAssets):
"source_run_id": "source_run_id_1",
"source_map_index": -1,
"source_task_instance_id": None,
+ "source_region_id": None,
+ "source_region_index": None,
"created_dagruns": [
{
"run_id": "source_run_id_1",
@@ -1508,6 +1560,8 @@ class TestGetAssetEvents(TestAssets):
"source_run_id": "source_run_id_2",
"source_map_index": -1,
"source_task_instance_id": None,
+ "source_region_id": None,
+ "source_region_index": None,
"created_dagruns": [
{
"run_id": "source_run_id_2",
@@ -2673,6 +2727,8 @@ class TestPostAssetEvents(TestAssets):
"source_run_id": None,
"source_map_index": -1,
"source_task_instance_id": None,
+ "source_region_id": None,
+ "source_region_index": None,
"created_dagruns": [],
"timestamp": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
"partition_key": None,
@@ -2773,6 +2829,8 @@ class TestPostAssetEvents(TestAssets):
"source_run_id": None,
"source_map_index": -1,
"source_task_instance_id": None,
+ "source_region_id": None,
+ "source_region_index": None,
"created_dagruns": [],
"timestamp": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
"partition_key": None,
@@ -2825,6 +2883,7 @@ class TestPostAssetEventsTeamResolution(TestAssets):
extra={"from_rest_api": True},
source_map_index=-1,
source_task_instance_id=None,
+ source_task_instance=None,
timestamp=DEFAULT_DATE,
source_task_id=None,
source_dag_id=None,
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
index d3c5900d4e5..569c9102265 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
@@ -2055,6 +2055,8 @@ class TestGetDagRunAssetTriggerEvents:
"source_dag_id": ti.dag_id,
"source_map_index": ti.map_index,
"source_task_instance_id": None,
+ "source_region_id": None,
+ "source_region_index": None,
"source_run_id": ti.run_id,
"source_task_id": ti.task_id,
"created_dagruns": [
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index 21d6850a4b0..4afb2311d34 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -1911,6 +1911,8 @@ class AssetEventResponse(BaseModel):
source_run_id: Annotated[str | None, Field(title="Source Run Id")] = None
source_map_index: Annotated[int, Field(title="Source Map Index")]
source_task_instance_id: Annotated[UUID | None, Field(title="Source Task
Instance Id")] = None
+ source_region_id: Annotated[UUID | None, Field(title="Source Region Id")]
= None
+ source_region_index: Annotated[int | None, Field(title="Source Region
Index")] = None
created_dagruns: Annotated[list[DagRunAssetReference],
Field(title="Created Dagruns")]
timestamp: Annotated[datetime, Field(title="Timestamp")]
partition_key: Annotated[str | None, Field(title="Partition Key")] = None
