This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 513e46edad5 UI: Show owning team in asset dependency popovers (#71446)
513e46edad5 is described below
commit 513e46edad5d89bbb6dbb541758f4257b31cde62
Author: Vincent <[email protected]>
AuthorDate: Wed Aug 19 11:43:56 2026 -0400
UI: Show owning team in asset dependency popovers (#71446)
Tracing data lineage across a multi-team deployment means jumping between
assets and the Dags and tasks that produce or schedule them, without any
signal of who owns each one. Surfacing the owning team directly in the
"Scheduled Dags" and "Producing Tasks" popovers on the assets list makes
ownership obvious while following those links, so operators know who to
reach without opening each Dag.
The team is resolved in a single batched lookup and only when multi-team
support is enabled, so single-team deployments issue no extra query and
see no change.
---
.../src/airflow/api_fastapi/common/db/assets.py | 28 +++++-
.../api_fastapi/core_api/datamodels/assets.py | 2 +
.../api_fastapi/core_api/openapi/_private_ui.yaml | 10 ++
.../core_api/openapi/v2-rest-api-generated.yaml | 10 ++
.../api_fastapi/core_api/routes/public/assets.py | 7 +-
airflow-core/src/airflow/models/asset.py | 18 ++++
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 22 +++++
.../airflow/ui/openapi-gen/requests/types.gen.ts | 2 +
.../pages/AssetsList/DependencyPopover.test.tsx | 102 +++++++++++++++++++++
.../ui/src/pages/AssetsList/DependencyPopover.tsx | 40 +++++++-
.../core_api/routes/public/test_assets.py | 85 +++++++++++++++++
.../api_fastapi/core_api/routes/ui/test_assets.py | 41 +++++++++
.../src/airflowctl/api/datamodels/generated.py | 2 +
13 files changed, 357 insertions(+), 12 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/db/assets.py
b/airflow-core/src/airflow/api_fastapi/common/db/assets.py
index 3e54c8dcabb..8ee707fc9cd 100644
--- a/airflow-core/src/airflow/api_fastapi/common/db/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/common/db/assets.py
@@ -23,11 +23,20 @@ from typing import TYPE_CHECKING
from sqlalchemy import func, select
from sqlalchemy.orm import selectinload
-from airflow.models.asset import AssetEvent, AssetModel, AssetWatcherModel,
association_table
+from airflow.api_fastapi.common.db.dags import eager_load_teams
+from airflow.models.asset import (
+ AssetEvent,
+ AssetModel,
+ AssetWatcherModel,
+ DagScheduleAssetReference,
+ TaskOutletAssetReference,
+ association_table,
+)
from airflow.models.dagrun import DagRun
if TYPE_CHECKING:
from sqlalchemy.orm import Session
+ from sqlalchemy.orm.strategy_options import _AbstractLoad
from sqlalchemy.sql import Select
@@ -48,8 +57,7 @@ def generate_assets_with_last_event_query() -> Select:
.outerjoin(max_asset_event_id_query, AssetModel.id ==
max_asset_event_id_query.c.asset_id)
.outerjoin(AssetEvent, AssetEvent.id ==
max_asset_event_id_query.c.max_asset_event_id)
.options(
- selectinload(AssetModel.scheduled_dags),
- selectinload(AssetModel.producing_tasks),
+ *eager_load_asset_reference_teams(),
selectinload(AssetModel.consuming_tasks),
selectinload(AssetModel.aliases),
selectinload(AssetModel.watchers).joinedload(AssetWatcherModel.trigger),
@@ -94,3 +102,17 @@ def resolve_triggering_event_refs(
)
)
}
+
+
+def eager_load_asset_reference_teams() -> tuple[_AbstractLoad, ...]:
+ """
+ Loader options for an ``AssetModel`` query whose response exposes
reference ``team_name``.
+
+ Covers the two references that serialize a team: scheduled Dags and
producing tasks.
+ The Dag each one points at is only traversed in multi-team mode, so
single-team
+ deployments load the references alone -- see :func:`eager_load_teams`.
+ """
+ return (
+
selectinload(AssetModel.scheduled_dags).options(*eager_load_teams(DagScheduleAssetReference.dag)),
+
selectinload(AssetModel.producing_tasks).options(*eager_load_teams(TaskOutletAssetReference.dag)),
+ )
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 d4bb4c2ce2b..2d5c18050cc 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
@@ -50,6 +50,7 @@ class DagScheduleAssetReference(StrictBaseModel):
dag_id: str
created_at: datetime
updated_at: datetime
+ team_name: str | None = None
class TaskInletAssetReference(StrictBaseModel):
@@ -68,6 +69,7 @@ class TaskOutletAssetReference(StrictBaseModel):
task_id: str
created_at: datetime
updated_at: datetime
+ team_name: str | None = None
class LastAssetEventResponse(BaseModel):
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
index 5abfa412e58..6a8af48ecf6 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
@@ -3154,6 +3154,11 @@ components:
type: string
format: date-time
title: Updated At
+ team_name:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Team Name
additionalProperties: false
type: object
required:
@@ -4712,6 +4717,11 @@ components:
type: string
format: date-time
title: Updated At
+ team_name:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Team Name
additionalProperties: false
type: object
required:
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 417162a70e1..ce8920b702d 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
@@ -14053,6 +14053,11 @@ components:
type: string
format: date-time
title: Updated At
+ team_name:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Team Name
additionalProperties: false
type: object
required:
@@ -16168,6 +16173,11 @@ components:
type: string
format: date-time
title: Updated At
+ team_name:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Team Name
additionalProperties: false
type: object
required:
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 5b7f5ec75b4..14c40480231 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
@@ -29,6 +29,7 @@ from airflow._shared.timezones import timezone
from airflow.api_fastapi.app import get_auth_manager
from airflow.api_fastapi.auth.managers.models.resource_details import
DagAccessEntity, DagDetails
from airflow.api_fastapi.common.dagbag import DagBagDep,
get_latest_version_of_dag
+from airflow.api_fastapi.common.db.assets import
eager_load_asset_reference_teams
from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
from airflow.api_fastapi.common.parameters import (
BaseParam,
@@ -211,8 +212,7 @@ def get_assets(
# The below type annotation is acceptable on SQLA2.1, but not on 2.0
assets_rows: Result[Unpack[tuple[AssetModel, int, datetime]]] =
session.execute( # type: ignore[type-arg]
assets_select.options(
- subqueryload(AssetModel.scheduled_dags),
- subqueryload(AssetModel.producing_tasks),
+ *eager_load_asset_reference_teams(),
subqueryload(AssetModel.consuming_tasks),
subqueryload(AssetModel.aliases),
subqueryload(AssetModel.watchers).joinedload(AssetWatcherModel.trigger),
@@ -590,8 +590,7 @@ def get_asset(
select(AssetModel)
.where(AssetModel.id == asset_id)
.options(
- joinedload(AssetModel.scheduled_dags),
- joinedload(AssetModel.producing_tasks),
+ *eager_load_asset_reference_teams(),
joinedload(AssetModel.consuming_tasks),
joinedload(AssetModel.watchers).joinedload(AssetWatcherModel.trigger),
)
diff --git a/airflow-core/src/airflow/models/asset.py
b/airflow-core/src/airflow/models/asset.py
index 750aac8d2b7..d980c2d69ac 100644
--- a/airflow-core/src/airflow/models/asset.py
+++ b/airflow-core/src/airflow/models/asset.py
@@ -39,6 +39,7 @@ from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm import Mapped, mapped_column, relationship
from airflow._shared.timezones import timezone
+from airflow.configuration import conf as airflow_conf
from airflow.models.base import Base, StringID
from airflow.utils.sqlalchemy import UtcDateTime
@@ -605,6 +606,14 @@ class DagScheduleAssetReference(Base):
asset = relationship("AssetModel", back_populates="scheduled_dags")
dag = relationship("DagModel", back_populates="schedule_asset_references")
+ @property
+ def team_name(self) -> str | None:
+ """Name of the team owning the Dag scheduled by this asset, or
``None``."""
+ # Gate before touching ``dag``: single-team deployments must not pay
for the load.
+ if not airflow_conf.getboolean("core", "multi_team"):
+ return None
+ return self.dag.team_name if self.dag else None
+
queue_records = relationship(
"AssetDagRunQueue",
primaryjoin="""and_(
@@ -661,6 +670,15 @@ class TaskOutletAssetReference(Base):
)
asset = relationship("AssetModel", back_populates="producing_tasks")
+ dag = relationship("DagModel", viewonly=True)
+
+ @property
+ def team_name(self) -> str | None:
+ """Name of the team owning the Dag producing this asset, or
``None``."""
+ # Gate before touching ``dag``: single-team deployments must not pay
for the load.
+ if not airflow_conf.getboolean("core", "multi_team"):
+ return None
+ return self.dag.team_name if self.dag else None
__tablename__ = "task_outlet_asset_reference"
__table_args__ = (
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 3225ccc4660..824a7f4d4b7 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
@@ -4564,6 +4564,17 @@ export const $DagScheduleAssetReference = {
type: 'string',
format: 'date-time',
title: 'Updated At'
+ },
+ team_name: {
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Team Name'
}
},
additionalProperties: false,
@@ -7740,6 +7751,17 @@ export const $TaskOutletAssetReference = {
type: 'string',
format: 'date-time',
title: 'Updated At'
+ },
+ team_name: {
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Team Name'
}
},
additionalProperties: false,
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 7fa6964faf1..c060e08aa19 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
@@ -1193,6 +1193,7 @@ export type DagScheduleAssetReference = {
dag_id: string;
created_at: string;
updated_at: string;
+ team_name?: string | null;
};
/**
@@ -1938,6 +1939,7 @@ export type TaskOutletAssetReference = {
task_id: string;
created_at: string;
updated_at: string;
+ team_name?: string | null;
};
/**
diff --git
a/airflow-core/src/airflow/ui/src/pages/AssetsList/DependencyPopover.test.tsx
b/airflow-core/src/airflow/ui/src/pages/AssetsList/DependencyPopover.test.tsx
new file mode 100644
index 00000000000..8ae2e3008f6
--- /dev/null
+++
b/airflow-core/src/airflow/ui/src/pages/AssetsList/DependencyPopover.test.tsx
@@ -0,0 +1,102 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import "@testing-library/jest-dom";
+import { fireEvent, render, screen, waitFor } from "@testing-library/react";
+import type { DagScheduleAssetReference, TaskOutletAssetReference } from
"openapi-gen/requests/types.gen";
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import i18n from "src/i18n/config";
+import { Wrapper } from "src/utils/Wrapper";
+
+import { DependencyPopover } from "./DependencyPopover";
+
+const mockConfig: Record<string, unknown> = { multi_team: false };
+
+vi.mock("src/queries/useConfig", () => ({
+ useConfig: (key: string) => mockConfig[key],
+}));
+
+const scheduledDags = [
+ {
+ created_at: "2024-01-01T00:00:00Z",
+ dag_id: "dag_a",
+ team_name: "team-a",
+ updated_at: "2024-01-01T00:00:00Z",
+ },
+] as Array<DagScheduleAssetReference>;
+
+const producingTasks = [
+ {
+ created_at: "2024-01-01T00:00:00Z",
+ dag_id: "dag_b",
+ task_id: "task_b",
+ team_name: "team-b",
+ updated_at: "2024-01-01T00:00:00Z",
+ },
+] as Array<TaskOutletAssetReference>;
+
+describe("DependencyPopover", () => {
+ afterEach(() => {
+ mockConfig.multi_team = false;
+ });
+
+ it("shows the owning team next to each scheduled Dag when multi-team is
enabled", async () => {
+ mockConfig.multi_team = true;
+ render(
+ <Wrapper>
+ <DependencyPopover dependencies={scheduledDags} type="Dag" />
+ </Wrapper>,
+ );
+
+ fireEvent.click(screen.getByRole("button"));
+
+ await waitFor(() => expect(screen.getByRole("link", { name: "dag_a"
})).toBeInTheDocument());
+
expect(screen.getByText(i18n.t("common:dagDetails.team"))).toBeInTheDocument();
+ expect(screen.getByRole("link", { name: "team-a"
})).toHaveAttribute("href", "/dags?teams=team-a");
+ });
+
+ it("shows the owning team next to each producing task when multi-team is
enabled", async () => {
+ mockConfig.multi_team = true;
+ render(
+ <Wrapper>
+ <DependencyPopover dependencies={producingTasks} type="Task" />
+ </Wrapper>,
+ );
+
+ fireEvent.click(screen.getByRole("button"));
+
+ await waitFor(() => expect(screen.getByRole("link", { name: "dag_b.task_b"
})).toBeInTheDocument());
+
expect(screen.getByText(i18n.t("common:dagDetails.team"))).toBeInTheDocument();
+ expect(screen.getByRole("link", { name: "team-b"
})).toHaveAttribute("href", "/dags?teams=team-b");
+ });
+
+ it("does not show the team when multi-team is disabled", async () => {
+ render(
+ <Wrapper>
+ <DependencyPopover dependencies={scheduledDags} type="Dag" />
+ </Wrapper>,
+ );
+
+ fireEvent.click(screen.getByRole("button"));
+
+ await waitFor(() => expect(screen.getByRole("link", { name: "dag_a"
})).toBeInTheDocument());
+
expect(screen.queryByText(i18n.t("common:dagDetails.team"))).not.toBeInTheDocument();
+ expect(screen.queryByText("team-a")).not.toBeInTheDocument();
+ });
+});
diff --git
a/airflow-core/src/airflow/ui/src/pages/AssetsList/DependencyPopover.tsx
b/airflow-core/src/airflow/ui/src/pages/AssetsList/DependencyPopover.tsx
index 44438e10fba..1576aae8a63 100644
--- a/airflow-core/src/airflow/ui/src/pages/AssetsList/DependencyPopover.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/AssetsList/DependencyPopover.tsx
@@ -16,17 +16,46 @@
* specific language governing permissions and limitations
* under the License.
*/
-import { Button } from "@chakra-ui/react";
+import { Button, HStack, Icon, VisuallyHidden } from "@chakra-ui/react";
import { useTranslation } from "react-i18next";
+import { FiUsers } from "react-icons/fi";
import type { DagScheduleAssetReference, TaskOutletAssetReference } from
"openapi/requests/types.gen";
-import { Popover, RouterLink } from "src/components/ui";
+import { TeamName } from "src/components/TeamName";
+import { Popover, RouterLink, Tooltip } from "src/components/ui";
+import { useShowTeam } from "src/hooks/useShowTeam";
type Props = {
readonly dependencies: Array<DagScheduleAssetReference |
TaskOutletAssetReference>;
readonly type: "Dag" | "Task";
};
+// Elsewhere a team sits under a "Team" column header or detail-row label; the
popover has no such
+// heading, so the icon marks the link as a team rather than a second Dag
link. ``FiUsers`` is the
+// same icon the teams filter uses.
+const IconTeamName = ({ teamName }: { readonly teamName?: string | null }) => {
+ const { t: translate } = useTranslation("common");
+ const showTeam = useShowTeam(teamName);
+
+ if (!showTeam) {
+ return undefined;
+ }
+
+ return (
+ <Tooltip content={translate("dagDetails.team")}>
+ <HStack gap={1}>
+ <Icon color="fg.muted">
+ <FiUsers />
+ </Icon>
+ {/* The icon is the only cue that this link is a team rather than
another Dag, and Chakra
+ hides icons from assistive tech, so the label is announced
separately. */}
+ <VisuallyHidden>{translate("dagDetails.team")}</VisuallyHidden>
+ <TeamName teamName={teamName} />
+ </HStack>
+ </Tooltip>
+ );
+};
+
export const DependencyPopover = ({ dependencies, type }: Props) => {
const { t: translate } = useTranslation("common");
const dependencyKey = type.toLowerCase() as "dag" | "task";
@@ -56,9 +85,10 @@ export const DependencyPopover = ({ dependencies, type }:
Props) => {
}
return (
- <RouterLink display="block" key={key} py={2} to={link}>
- {label}
- </RouterLink>
+ <HStack gap={2} justifyContent="space-between" key={key} py={2}>
+ <RouterLink to={link}>{label}</RouterLink>
+ <IconTeamName teamName={dependency.team_name} />
+ </HStack>
);
})}
</Popover.Body>
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 d3a94af6ba0..1ae3690e19d 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
@@ -40,8 +40,10 @@ from airflow.models.asset import (
TaskOutletAssetReference,
)
from airflow.models.base import ID_LEN
+from airflow.models.dagbundle import DagBundleModel
from airflow.models.dagrun import DagRun
from airflow.models.serialized_dag import SerializedDagModel
+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
@@ -59,6 +61,7 @@ from tests_common.test_utils.db import (
clear_db_dags,
clear_db_logs,
clear_db_runs,
+ clear_db_teams,
)
from tests_common.test_utils.format_datetime import
from_datetime_to_zulu_without_ms
from tests_common.test_utils.logs import check_last_log
@@ -136,6 +139,30 @@ def _create_assets_with_watchers(session, num: int = 2) ->
list[AssetModel]:
return assets
+def _create_assets_with_team_references(session, num: int = 2, refs_per_asset:
int = 1) -> list[AssetModel]:
+ """Create ``num`` assets, each scheduling and produced by
``refs_per_asset`` team-owned Dags."""
+ bundle = DagBundleModel(name="team-bundle-assets")
+ bundle.teams.append(Team(name="team-assets"))
+ session.add(bundle)
+ session.flush()
+ assets = [AssetModel(name=f"asset{i}", uri=f"s3://bucket/asset{i}",
group="asset") for i in range(num)]
+ session.add_all(assets)
+ session.add_all(AssetActive.for_asset(asset) for asset in assets)
+ session.flush()
+ for i, asset in enumerate(assets):
+ for j in range(refs_per_asset):
+ session.add_all(
+ [
+ DagModel(dag_id=f"scheduled_dag{i}_{j}",
bundle_name="team-bundle-assets"),
+ DagModel(dag_id=f"producing_dag{i}_{j}",
bundle_name="team-bundle-assets"),
+ DagScheduleAssetReference(dag_id=f"scheduled_dag{i}_{j}",
asset=asset),
+ TaskOutletAssetReference(dag_id=f"producing_dag{i}_{j}",
task_id="task1", asset=asset),
+ ]
+ )
+ session.commit()
+ return assets
+
+
def _create_assets_with_sensitive_extra(session, num: int = 2) -> None:
assets = [
AssetModel(
@@ -292,6 +319,7 @@ class TestAssets:
clear_db_assets()
clear_db_runs()
clear_db_dags()
+ clear_db_teams()
clear_db_dag_bundles()
clear_db_logs()
@@ -528,6 +556,45 @@ class TestGetAssets(TestAssets):
msg = "Ordering with 'fake' is disallowed or the attribute does not
exist on the model"
assert response.json()["detail"] == msg
+ def test_assets_references_team_name_none_without_multi_team(self,
test_client, session):
+ """Without multi-team enabled, references keep ``team_name`` of
``None`` and no lookup happens."""
+ _create_assets_with_team_references(session)
+
+ response = test_client.get("/assets")
+ assert response.status_code == 200
+ assets = {asset["name"]: asset for asset in response.json()["assets"]}
+ assert assets["asset0"]["scheduled_dags"][0]["team_name"] is None
+ assert assets["asset0"]["producing_tasks"][0]["team_name"] is None
+
+ @conf_vars({("core", "multi_team"): "True"})
+ def test_assets_references_include_team_name(self, test_client, session):
+ """With multi-team enabled, the owning team is attached to scheduled
Dags and producing tasks."""
+ _create_assets_with_team_references(session)
+
+ response = test_client.get("/assets")
+ assert response.status_code == 200
+ assets = {asset["name"]: asset for asset in response.json()["assets"]}
+ assert assets["asset0"]["scheduled_dags"][0]["team_name"] ==
"team-assets"
+ assert assets["asset0"]["producing_tasks"][0]["team_name"] ==
"team-assets"
+
+ @conf_vars({("core", "multi_team"): "True"})
+ def test_query_count_with_multi_team(self, test_client, session):
+ """Resolving reference ``team_name`` must not add a query per
referencing Dag.
+
+ A missing loader option does not raise: :attr:`DagModel.team_name`
falls back to the
+ cached ``get_team_name`` resolver instead of tripping
``lazy="raise"``, so only a pinned
+ count catches the regression.
+ """
+ _create_assets_with_team_references(session, num=5)
+
+ with assert_queries_count(9):
+ response = test_client.get("/assets")
+
+ assert response.status_code == 200
+ assets = {asset["name"]: asset for asset in response.json()["assets"]}
+ assert assets["asset4"]["scheduled_dags"][0]["team_name"] ==
"team-assets"
+ assert assets["asset4"]["producing_tasks"][0]["team_name"] ==
"team-assets"
+
@pytest.mark.parametrize(
("params", "expected_assets"),
[
@@ -1617,6 +1684,24 @@ class TestGetAssetEndpoint(TestAssets):
"last_asset_event": {"id": None, "timestamp": None},
}
+ @conf_vars({("core", "multi_team"): "True"})
+ def test_query_count_with_multi_team(self, test_client, session):
+ """Resolving reference ``team_name`` must not add a query per
referencing Dag.
+
+ A missing loader option does not raise: :attr:`DagModel.team_name`
falls back to the
+ cached ``get_team_name`` resolver instead of tripping
``lazy="raise"``, so only a pinned
+ count catches the regression.
+ """
+ asset = _create_assets_with_team_references(session, num=1,
refs_per_asset=5)[0]
+
+ with assert_queries_count(8):
+ response = test_client.get(f"/assets/{asset.id}")
+
+ assert response.status_code == 200
+ body = response.json()
+ assert {ref["team_name"] for ref in body["scheduled_dags"]} ==
{"team-assets"}
+ assert {ref["team_name"] for ref in body["producing_tasks"]} ==
{"team-assets"}
+
def test_should_respond_401(self, unauthenticated_test_client):
response = unauthenticated_test_client.get("/assets/1")
assert response.status_code == 401
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
index 9cee5f1ac8c..f91f475a18b 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
@@ -34,7 +34,10 @@ from airflow.models.asset import (
AssetWatcherModel,
DagScheduleAssetReference,
PartitionedAssetKeyLog,
+ TaskOutletAssetReference,
)
+from airflow.models.dagbundle import DagBundleModel
+from airflow.models.team import Team
from airflow.models.trigger import Trigger
from airflow.partition_mappers.base import RollupMapper
from airflow.partition_mappers.temporal import StartOfHourMapper
@@ -44,12 +47,15 @@ from airflow.sdk.definitions.asset import Asset
from airflow.sdk.definitions.timetables.assets import PartitionedAssetTimetable
from tests_common.test_utils.asserts import assert_queries_count
+from tests_common.test_utils.config import conf_vars
from tests_common.test_utils.db import (
clear_db_apdr,
clear_db_assets,
+ clear_db_dag_bundles,
clear_db_dags,
clear_db_pakl,
clear_db_serialized_dags,
+ clear_db_teams,
)
pytestmark = pytest.mark.db_test
@@ -61,6 +67,8 @@ def cleanup():
clear_db_serialized_dags()
clear_db_apdr()
clear_db_pakl()
+ clear_db_teams()
+ clear_db_dag_bundles()
class TestNextRunAssets:
@@ -778,3 +786,36 @@ class TestGetAssetsUi:
with assert_queries_count(8):
assert test_client.get("/assets").status_code == 200
+
+ @conf_vars({("core", "multi_team"): "True"})
+ def test_query_count_with_multi_team(self, test_client, session):
+ """Resolving reference ``team_name`` must not add a query per
referencing Dag.
+
+ A missing loader option does not raise: :attr:`DagModel.team_name`
falls back to the
+ cached ``get_team_name`` resolver instead of tripping
``lazy="raise"``, so only a pinned
+ count catches the regression.
+ """
+ bundle = DagBundleModel(name="team-bundle")
+ bundle.teams.append(Team(name="owning-team"))
+ session.add(bundle)
+ session.flush()
+
+ assets = [AssetModel(name=f"asset{i}", uri=f"s3://bucket/asset{i}",
group="asset") for i in range(5)]
+ session.add_all(assets)
+ session.add_all(AssetActive.for_asset(asset) for asset in assets)
+ session.flush()
+
+ for i, asset in enumerate(assets):
+ session.add(DagModel(dag_id=f"scheduled_dag{i}",
bundle_name="team-bundle"))
+ session.add(DagModel(dag_id=f"producing_dag{i}",
bundle_name="team-bundle"))
+ session.add(DagScheduleAssetReference(dag_id=f"scheduled_dag{i}",
asset=asset))
+ session.add(TaskOutletAssetReference(dag_id=f"producing_dag{i}",
task_id="task", asset=asset))
+ session.commit()
+
+ with assert_queries_count(12):
+ response = test_client.get("/assets")
+
+ assert response.status_code == 200
+ assets_by_name = {asset["name"]: asset for asset in
response.json()["assets"]}
+ assert assets_by_name["asset0"]["scheduled_dags"][0]["team_name"] ==
"owning-team"
+ assert assets_by_name["asset0"]["producing_tasks"][0]["team_name"] ==
"owning-team"
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index 0dc38fa37dc..c057f5548c9 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -606,6 +606,7 @@ class DagScheduleAssetReference(BaseModel):
dag_id: Annotated[str, Field(title="Dag Id")]
created_at: Annotated[datetime, Field(title="Created At")]
updated_at: Annotated[datetime, Field(title="Updated At")]
+ team_name: Annotated[str | None, Field(title="Team Name")] = None
class DagStatsStateResponse(BaseModel):
@@ -1100,6 +1101,7 @@ class TaskOutletAssetReference(BaseModel):
task_id: Annotated[str, Field(title="Task Id")]
created_at: Annotated[datetime, Field(title="Created At")]
updated_at: Annotated[datetime, Field(title="Updated At")]
+ team_name: Annotated[str | None, Field(title="Team Name")] = None
class TaskStateStoreBody(BaseModel):
