This is an automated email from the ASF dual-hosted git repository. vatsrahul1001 pushed a commit to branch backport-230cd8a-v3-4-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit f7cf9f16617f5c868379a5ad4eae3be914564ce6 Author: Brent Bovenzi <[email protected]> AuthorDate: Tue Oct 13 14:50:50 2026 -0400 Let plugin views be scoped to specific Assets (#74609) Asset pages could not be scoped at all. `asset` resolved no record, so every `applies_to` path on an asset view was unevaluable and skipped, and skipping does not narrow anything -- an asset view configured with scoping showed on every asset, which is the opposite of what its author asked for. The asset is already fetched for those pages, so it costs nothing to make it a root an `applies_to` path can name, alongside the Dag, Dag Run, Task and Task Instance records. `AssetResponse` joins the models paths are validated against, so an asset path gets the same misspelling check as any other. (cherry picked from commit 230cd8adc33bc4e693e875d059422a5875cbc142) --- .../docs/administration-and-deployment/plugins.rst | 7 ++- airflow-core/src/airflow/plugins_manager.py | 7 ++- .../ui/src/hooks/usePluginAppliesToContext.test.ts | 21 +++++++++ .../ui/src/hooks/usePluginAppliesToContext.ts | 14 ++++-- .../airflow/ui/src/utils/pluginAppliesTo.test.ts | 21 +++++++++ .../src/airflow/ui/src/utils/pluginAppliesTo.ts | 6 ++- .../tests/unit/plugins/test_plugins_manager.py | 53 ++++++++++++++++++++++ 7 files changed, 121 insertions(+), 8 deletions(-) diff --git a/airflow-core/docs/administration-and-deployment/plugins.rst b/airflow-core/docs/administration-and-deployment/plugins.rst index d5861d0bb17..b9965b54349 100644 --- a/airflow-core/docs/administration-and-deployment/plugins.rst +++ b/airflow-core/docs/administration-and-deployment/plugins.rst @@ -420,7 +420,7 @@ and compared as strings, so numbers and booleans need no quoting rules of their An **unqualified path is rooted at the entity the destination is about** — on ``dag_run``, ``state`` is the run's state; on ``task_instance``, the task instance's. A path may instead -name a related record as its first segment: ``dag``, ``dag_run``, ``task`` or +name a related record as its first segment: ``asset``, ``dag``, ``dag_run``, ``task`` or ``task_instance``. Traversing a list fans out across it, so ``dag.tags.name`` collects every tag name and matches if any of them is listed. @@ -448,7 +448,10 @@ resolves: * - ``task_instance`` - ``task_instance`` - ``dag``, ``dag_run``, ``task``, ``task_instance`` - * - ``nav``, ``base``, ``dashboard``, ``asset`` + * - ``asset`` + - ``asset`` + - ``asset`` + * - ``nav``, ``base``, ``dashboard`` - — - none, so every path is skipped diff --git a/airflow-core/src/airflow/plugins_manager.py b/airflow-core/src/airflow/plugins_manager.py index 67021ef58ae..a2edbab82d1 100644 --- a/airflow-core/src/airflow/plugins_manager.py +++ b/airflow-core/src/airflow/plugins_manager.py @@ -174,11 +174,12 @@ _APPLIES_TO_ROOTS: dict[str, frozenset[str]] = { "nav": frozenset(), "base": frozenset(), "dashboard": frozenset(), - "asset": frozenset(), + "asset": frozenset({"asset"}), } # Which record an unqualified path is rooted at -- the entity the destination is about. _APPLIES_TO_ENTITY_ROOT: dict[str, str] = { + "asset": "asset", "dag": "dag", "dag_overview": "dag", "dag_run": "dag_run", @@ -187,7 +188,7 @@ _APPLIES_TO_ENTITY_ROOT: dict[str, str] = { "task_instance": "task_instance", } -_APPLIES_TO_ROOT_NAMES = frozenset({"dag", "dag_run", "task", "task_instance"}) +_APPLIES_TO_ROOT_NAMES = frozenset({"asset", "dag", "dag_run", "task", "task_instance"}) def _applies_to_path_root(path: str, destination: str) -> str | None: @@ -220,12 +221,14 @@ def _applies_to_root_models() -> dict[str, Any]: These are the models the UI actually fetches for the ``applies_to`` context -- keep them in step with ``AppliesToContext`` in ``src/utils/pluginAppliesTo.ts``. """ + from airflow.api_fastapi.core_api.datamodels.assets import AssetResponse from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse from airflow.api_fastapi.core_api.datamodels.task_instances import TaskInstanceResponse from airflow.api_fastapi.core_api.datamodels.tasks import TaskResponse return { + "asset": AssetResponse, "dag": DAGResponse, "dag_run": DAGRunResponse, "task": TaskResponse, diff --git a/airflow-core/src/airflow/ui/src/hooks/usePluginAppliesToContext.test.ts b/airflow-core/src/airflow/ui/src/hooks/usePluginAppliesToContext.test.ts index 2231eb17ec7..3c4be623cbd 100644 --- a/airflow-core/src/airflow/ui/src/hooks/usePluginAppliesToContext.test.ts +++ b/airflow-core/src/airflow/ui/src/hooks/usePluginAppliesToContext.test.ts @@ -61,6 +61,7 @@ const { calls, data, record } = vi.hoisted(() => { vi.mock("openapi/queries", async (importOriginal) => ({ ...(await importOriginal<typeof OpenapiQueries>()), + useAssetServiceGetAsset: record("asset"), useDagRunServiceGetDagRun: record("dagRun"), useDagServiceGetDag: record("dag"), useTaskInstanceServiceGetMappedTaskInstance: record("taskInstance"), @@ -82,6 +83,7 @@ describe("usePluginAppliesToContext", () => { it("issues no query when no view needs scoping", () => { renderContext(false); + expect(calls.asset?.options?.enabled).toBe(false); expect(calls.dag?.options?.enabled).toBe(false); expect(calls.dagRun?.options?.enabled).toBe(false); expect(calls.task?.options?.enabled).toBe(false); @@ -191,6 +193,25 @@ describe("usePluginAppliesToContext", () => { expect(calls.taskInstance?.options?.enabled).toBe(false); }); + it("resolves the asset on an asset route, and nothing else", () => { + mockParams = { assetId: "7" }; + + renderContext(true); + + expect(calls.asset?.options?.enabled).toBe(true); + expect(calls.asset?.params).toStrictEqual({ assetId: 7 }); + expect(calls.dag?.options?.enabled).toBe(false); + expect(calls.dagRun?.options?.enabled).toBe(false); + }); + + it("leaves the asset query disabled off an asset route", () => { + mockParams = { dagId }; + + renderContext(true); + + expect(calls.asset?.options?.enabled).toBe(false); + }); + it("resolves only the Dag on a Dag-level route", () => { mockParams = { dagId }; diff --git a/airflow-core/src/airflow/ui/src/hooks/usePluginAppliesToContext.ts b/airflow-core/src/airflow/ui/src/hooks/usePluginAppliesToContext.ts index b95afb7669d..a0c8f048baa 100644 --- a/airflow-core/src/airflow/ui/src/hooks/usePluginAppliesToContext.ts +++ b/airflow-core/src/airflow/ui/src/hooks/usePluginAppliesToContext.ts @@ -19,6 +19,7 @@ import { useParams } from "react-router-dom"; import { + useAssetServiceGetAsset, useDagRunServiceGetDagRun, useDagServiceGetDag, useTaskInstanceServiceGetMappedTaskInstance, @@ -44,10 +45,16 @@ import type { AppliesToContext } from "src/utils/pluginAppliesTo"; * Task groups are skipped for the task query, since `groupId` is not a task_id and would 404. */ export const usePluginAppliesToContext = (enabled: boolean): AppliesToContext => { - const { dagId = "", groupId, mapIndex = "-1", runId = "", taskId = "" } = useParams(); + const { assetId, dagId = "", groupId, mapIndex = "-1", runId = "", taskId = "" } = useParams(); const coordinates = useTaskInstanceCoordinates(); const parsedMapIndex = parseInt(mapIndex, 10); + const { data: asset, isLoading: isAssetLoading } = useAssetServiceGetAsset( + { assetId: assetId === undefined ? 0 : parseInt(assetId, 10) }, + undefined, + { enabled: enabled && Boolean(assetId) }, + ); + const { data: dag, isLoading: isDagLoading } = useDagServiceGetDag({ dagId }, undefined, { enabled: enabled && Boolean(dagId), }); @@ -90,12 +97,13 @@ export const usePluginAppliesToContext = (enabled: boolean): AppliesToContext => ); return { + asset, dag, dagRun, // `isLoading` (not `isPending`) is deliberate: a disabled query reports // `isPending` forever, which would withhold scoped views indefinitely on - // destinations that legitimately have no run, task or task instance. - isLoading: isDagLoading || isDagRunLoading || isTaskLoading || isTaskInstanceLoading, + // destinations that legitimately have no asset, run, task or task instance. + isLoading: isAssetLoading || isDagLoading || isDagRunLoading || isTaskLoading || isTaskInstanceLoading, task, taskInstance, }; diff --git a/airflow-core/src/airflow/ui/src/utils/pluginAppliesTo.test.ts b/airflow-core/src/airflow/ui/src/utils/pluginAppliesTo.test.ts index 136c84d1166..b3bc2754d67 100644 --- a/airflow-core/src/airflow/ui/src/utils/pluginAppliesTo.test.ts +++ b/airflow-core/src/airflow/ui/src/utils/pluginAppliesTo.test.ts @@ -19,6 +19,7 @@ import { describe, expect, it } from "vitest"; import type { + AssetResponse, DAGResponse, DAGRunResponse, ExternalViewResponse, @@ -36,6 +37,9 @@ import { // These fixtures carry only the fields the tests address, so they are cast through `unknown` // rather than spelling out every field of the full response types. +const makeAsset = (name: string, group: string): AssetResponse => + ({ group, id: 1, name, uri: `s3://bucket/${name}` }) as unknown as AssetResponse; + const makeDag = (dagId: string, tagNames: Array<string>): DAGResponse => ({ dag_id: dagId, @@ -78,6 +82,7 @@ const makeView = ( const dag = makeDag("etl_sales", ["ml", "prod"]); +const assetContext: AppliesToContext = { asset: makeAsset("orders", "sales"), isLoading: false }; const dagContext: AppliesToContext = { dag, isLoading: false }; const dagRunContext: AppliesToContext = { dag, dagRun: makeDagRun("failed"), isLoading: false }; const taskContext: AppliesToContext = { @@ -167,6 +172,22 @@ describe("matchesAppliesTo — qualified paths", () => { }); }); +describe("matchesAppliesTo — asset destinations", () => { + it("roots an unqualified path at the asset", () => { + expect(matchesAppliesTo(makeView({ group: ["sales"] }, "asset"), assetContext)).toBe(true); + expect(matchesAppliesTo(makeView({ group: ["finance"] }, "asset"), assetContext)).toBe(false); + }); + + it("reaches the asset by naming it", () => { + expect(matchesAppliesTo(makeView({ "asset.name": ["orders"] }, "asset"), assetContext)).toBe(true); + expect(matchesAppliesTo(makeView({ "asset.name": ["refunds"] }, "asset"), assetContext)).toBe(false); + }); + + it("skips an asset path on a page that has no asset", () => { + expect(matchesAppliesTo(makeView({ "asset.name": ["orders"] }, "dag"), dagContext)).toBe(true); + }); +}); + describe("matchesAppliesTo — the three verdicts", () => { it("skips a path whose root record the surface does not have", () => { // A task-instance path on a Dag page: unevaluable, so it must not fail the match. diff --git a/airflow-core/src/airflow/ui/src/utils/pluginAppliesTo.ts b/airflow-core/src/airflow/ui/src/utils/pluginAppliesTo.ts index 179a14f3502..4298d49f213 100644 --- a/airflow-core/src/airflow/ui/src/utils/pluginAppliesTo.ts +++ b/airflow-core/src/airflow/ui/src/utils/pluginAppliesTo.ts @@ -17,6 +17,7 @@ * under the License. */ import type { + AssetResponse, DAGResponse, DAGRunResponse, ExternalViewResponse, @@ -35,6 +36,7 @@ export type PluginView = ExternalViewResponse | ReactAppResponse; * the route *does* have is still being fetched. */ export type AppliesToContext = { + asset?: AssetResponse; dag?: DAGResponse; dagRun?: DAGRunResponse; isLoading: boolean; @@ -42,11 +44,12 @@ export type AppliesToContext = { taskInstance?: TaskInstanceResponse; }; -type RootName = "dag" | "dagRun" | "task" | "taskInstance"; +type RootName = "asset" | "dag" | "dagRun" | "task" | "taskInstance"; // A path may name a related record as its first segment. Keyed by the wire spelling, since // that is what a plugin author writes. const ROOT_BY_PREFIX: Record<string, RootName> = { + asset: "asset", dag: "dag", dag_run: "dagRun", task: "task", @@ -57,6 +60,7 @@ const ROOT_BY_PREFIX: Record<string, RootName> = { // destination missing here can evaluate nothing, so all of its paths are skipped. // Kept in sync with `_APPLIES_TO_ENTITY_ROOT` in `airflow/plugins_manager.py`. const ENTITY_ROOT_BY_DESTINATION: Record<string, RootName> = { + asset: "asset", dag: "dag", dag_overview: "dag", dag_run: "dagRun", diff --git a/airflow-core/tests/unit/plugins/test_plugins_manager.py b/airflow-core/tests/unit/plugins/test_plugins_manager.py index 686caacd55d..e994baa923b 100644 --- a/airflow-core/tests/unit/plugins/test_plugins_manager.py +++ b/airflow-core/tests/unit/plugins/test_plugins_manager.py @@ -608,6 +608,59 @@ class TestPluginsManager: assert ExternalViewResponse(**external_views[0]).applies_to.root == {"dag.tags.name": ["ml"]} + def test_scopes_an_asset_view_on_the_asset_record(self, caplog): + class TestPlugin(AirflowPlugin): + name = "test_plugin" + + external_views = [ + { + "name": "Asset lineage", + "href": "/lineage", + "url_route": "/lineage", + "destination": "asset", + "applies_to": {"group": ["sales"], "asset.name": ["orders"]}, + } + ] + + with ( + mock_plugin_manager(plugins=[TestPlugin()]), + caplog.at_level(logging.WARNING, logger="airflow.plugins_manager"), + ): + from airflow import plugins_manager + + external_views, _ = plugins_manager._get_ui_plugins() + + assert external_views[0]["applies_to"] == {"group": ["sales"], "asset.name": ["orders"]} + + assert caplog.record_tuples == [] + + def test_withholds_an_asset_view_whose_path_matches_no_field(self, caplog): + class TestPlugin(AirflowPlugin): + name = "test_plugin" + + external_views = [ + { + "name": "Asset lineage", + "href": "/lineage", + "url_route": "/lineage", + "destination": "asset", + "applies_to": {"nme": ["orders"]}, + } + ] + + with ( + mock_plugin_manager(plugins=[TestPlugin()]), + caplog.at_level(logging.WARNING, logger="airflow.plugins_manager"), + ): + from airflow import plugins_manager + + external_views, _ = plugins_manager._get_ui_plugins() + + assert external_views == [] + + assert caplog.record_tuples[0][1] == logging.ERROR + assert "names no field 'nme' on AssetResponse" in caplog.record_tuples[0][2] + def test_accepts_a_path_through_a_field_the_models_do_not_describe(self, caplog): """A path cannot be checked past a bare ``dict``, so everything below it is accepted."""
