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 230cd8adc33 Let plugin views be scoped to specific Assets (#74609)
230cd8adc33 is described below
commit 230cd8adc33bc4e693e875d059422a5875cbc142
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.
---
.../docs/administration-and-deployment/plugins.rst | 7 ++-
airflow-core/newsfragments/74278.feature.rst | 2 +-
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 ++++++++++++++++++++++
8 files changed, 122 insertions(+), 9 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/newsfragments/74278.feature.rst
b/airflow-core/newsfragments/74278.feature.rst
index b6fba929773..04167d5bf45 100644
--- a/airflow-core/newsfragments/74278.feature.rst
+++ b/airflow-core/newsfragments/74278.feature.rst
@@ -1 +1 @@
-Plugin external views and React apps can be scoped with an optional
``applies_to`` block, a map of dotted field path to the values that path may
take, such as ``{"state": ["failed"], "dag.tags.name": ["ml"]}``. This replaces
the five fixed criteria added earlier in this development cycle and never
released (``dag_tags``, ``dag_ids``, ``task_ids``, ``operators``,
``operator_names``). A path naming a field no record has is treated as a
misconfiguration rather than a filter that never match [...]
+Plugin external views and React apps can be scoped with an optional
``applies_to`` block, a map of dotted field path to the values that path may
take, such as ``{"state": ["failed"], "dag.tags.name": ["ml"]}``. This replaces
the five fixed criteria added earlier in this development cycle and never
released (``dag_tags``, ``dag_ids``, ``task_ids``, ``operators``,
``operator_names``). Dag, Dag Run, Task, Task Instance and Asset pages can each
be scoped this way. A path naming a field no re [...]
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."""
