You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ra...@apache.org on 2026/04/23 05:03:48 UTC
(airflow) 08/48: Add dag runs filters (Consuming Asset) (#63624) (#65306)
This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 71d7b34e09bdd0c75683265e227e6858d3e71d67
Author: Rahul Vats <43...@users.noreply.github.com>
AuthorDate: Wed Apr 15 19:43:27 2026 +0530
Add dag runs filters (Consuming Asset) (#63624) (#65306)
* Add dag runs filters (Consuming Asset)
* Fix: correct consuming asset filter setup using association_table
* Trigger CI rebuild
* Rename consuming_asset filter to consuming_asset_pattern with database icon
* Rename consuming_asset filter to consuming_asset_pattern with database icon
* Trigger CI rebuild
* Fix consuming_asset_pattern naming
* Fix: rename consuming_asset to consuming_asset_pattern
* Fix: rename consuming_asset to consuming_asset_pattern
* Fix: Resolve PostgreSQL JSON comparison error in _ConsumingAssetFilter
* Rebase and fix _ConsumingAssetFilter
* Trigger CI
* add consumingAsset and filters.searchAsset to en/common.json
---------
(cherry picked from commit 5245419267b5892e7ff2e81042e1c4e21b48d6ca)
Co-authored-by: fat-catTW <12...@users.noreply.github.com>
Co-authored-by: Jens Scheffler <95...@users.noreply.github.com>
Co-authored-by: Jarek Potiuk <ja...@potiuk.com>
---
.../src/airflow/api_fastapi/common/parameters.py | 42 ++++++++++++++++++++
.../core_api/openapi/v2-rest-api-generated.yaml | 10 +++++
.../api_fastapi/core_api/routes/public/dag_run.py | 3 ++
.../src/airflow/ui/openapi-gen/queries/common.ts | 5 ++-
.../ui/openapi-gen/queries/ensureQueryData.ts | 6 ++-
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 6 ++-
.../src/airflow/ui/openapi-gen/queries/queries.ts | 6 ++-
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 6 ++-
.../ui/openapi-gen/requests/services.gen.ts | 4 +-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 4 ++
.../airflow/ui/public/i18n/locales/en/common.json | 2 +
.../src/airflow/ui/src/constants/filterConfigs.tsx | 9 ++++-
.../src/airflow/ui/src/constants/searchParams.ts | 1 +
airflow-core/src/airflow/ui/src/pages/DagRuns.tsx | 3 ++
.../src/airflow/ui/src/pages/DagRunsFilters.tsx | 1 +
.../src/airflow/ui/src/utils/useFiltersHandler.ts | 1 +
.../core_api/routes/public/test_dag_run.py | 45 ++++++++++++++++++++++
17 files changed, 142 insertions(+), 12 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters.py b/airflow-core/src/airflow/api_fastapi/common/parameters.py
index e963cdbff55..065dff8cda8 100644
--- a/airflow-core/src/airflow/api_fastapi/common/parameters.py
+++ b/airflow-core/src/airflow/api_fastapi/common/parameters.py
@@ -46,11 +46,13 @@ from airflow.configuration import conf
from airflow.models import Base
from airflow.models.asset import (
AssetAliasModel,
+ AssetEvent,
AssetModel,
AssetPartitionDagRun,
DagScheduleAssetReference,
TaskInletAssetReference,
TaskOutletAssetReference,
+ association_table,
)
from airflow.models.connection import Connection
from airflow.models.dag import DagModel, DagTag
@@ -785,6 +787,46 @@ QueryHasAssetScheduleFilter = Annotated[_HasAssetScheduleFilter, Depends(_HasAss
QueryAssetDependencyFilter = Annotated[_AssetDependencyFilter, Depends(_AssetDependencyFilter.depends)]
+class _ConsumingAssetFilter(BaseParam[str | None]):
+ """Filter DAG runs by consuming asset (name or URI)."""
+
+ def to_orm(self, select: Select) -> Select:
+ if not self.value and self.skip_none:
+ return select
+
+ event_subquery = (
+ sql_select(AssetEvent.id)
+ .join(AssetModel, AssetEvent.asset_id == AssetModel.id)
+ .where(
+ or_(
+ AssetModel.name.ilike(f"%{self.value}%"),
+ AssetModel.uri.ilike(f"%{self.value}%"),
+ )
+ )
+ .distinct()
+ )
+
+ dagrun_subquery = (
+ sql_select(association_table.c.dag_run_id)
+ .where(association_table.c.event_id.in_(event_subquery))
+ .distinct()
+ )
+
+ return select.where(DagRun.id.in_(dagrun_subquery))
+
+ @classmethod
+ def depends(
+ cls,
+ consuming_asset_pattern: str | None = Query(
+ None, description="Filter by consuming asset name or URI using pattern matching"
+ ),
+ ) -> _ConsumingAssetFilter:
+ return cls().set_value(consuming_asset_pattern)
+
+
+QueryConsumingAssetPatternSearch = Annotated[_ConsumingAssetFilter, Depends(_ConsumingAssetFilter.depends)]
+
+
class _PendingActionsFilter(BaseParam[bool]):
"""Filter Dags by having pending HITL actions (more than 1)."""
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 582357bbfea..6ed47e9fc2f 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
@@ -2375,6 +2375,16 @@ paths:
description: "SQL LIKE expression \u2014 use `%` / `_` wildcards (e.g. `%customer_%`).\
\ or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions\
\ are **not** supported."
+ - name: consuming_asset_pattern
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ description: Filter by consuming asset name or URI using pattern matching
+ title: Consuming Asset Pattern
+ description: Filter by consuming asset name or URI using pattern matching
responses:
'200':
description: Successful Response
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 ff42238806b..46dcd64a66e 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
@@ -46,6 +46,7 @@ from airflow.api_fastapi.common.parameters import (
FilterParam,
LimitFilter,
OffsetFilter,
+ QueryConsumingAssetPatternSearch,
QueryDagRunPartitionKeySearch,
QueryDagRunRunTypesFilter,
QueryDagRunStateFilter,
@@ -385,6 +386,7 @@ def get_dag_runs(
],
dag_id_pattern: Annotated[_SearchParam, Depends(search_param_factory(DagRun.dag_id, "dag_id_pattern"))],
partition_key_pattern: QueryDagRunPartitionKeySearch,
+ consuming_asset_pattern: QueryConsumingAssetPatternSearch,
) -> DAGRunCollectionResponse:
"""
Get all DAG Runs.
@@ -420,6 +422,7 @@ def get_dag_runs(
triggering_user_name_pattern,
dag_id_pattern,
partition_key_pattern,
+ consuming_asset_pattern,
],
order_by=order_by,
offset=offset,
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
index 612a3d56747..c02f01452b3 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -143,9 +143,10 @@ export const UseDagRunServiceGetUpstreamAssetEventsKeyFn = ({ dagId, dagRunId }:
export type DagRunServiceGetDagRunsDefaultResponse = Awaited<ReturnType<typeof DagRunService.getDagRuns>>;
export type DagRunServiceGetDagRunsQueryResult<TData = DagRunServiceGetDagRunsDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
export const useDagRunServiceGetDagRunsKey = "DagRunServiceGetDagRuns";
-export const UseDagRunServiceGetDagRunsKeyFn = ({ bundleVersion, confContains, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, startDateLte, state, triggeringUserNamePattern, updatedAtGt, upd [...]
+export const UseDagRunServiceGetDagRunsKeyFn = ({ bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, startDateLte, state, triggeringUserNamePa [...]
bundleVersion?: string;
confContains?: string;
+ consumingAssetPattern?: string;
dagId: string;
dagIdPattern?: string;
dagVersion?: number[];
@@ -181,7 +182,7 @@ export const UseDagRunServiceGetDagRunsKeyFn = ({ bundleVersion, confContains, d
updatedAtGte?: string;
updatedAtLt?: string;
updatedAtLte?: string;
-}, queryKey?: Array<unknown>) => [useDagRunServiceGetDagRunsKey, ...(queryKey ?? [{ bundleVersion, confContains, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, startDateLte, state, triggerin [...]
+}, queryKey?: Array<unknown>) => [useDagRunServiceGetDagRunsKey, ...(queryKey ?? [{ bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, startDa [...]
export type DagRunServiceWaitDagRunUntilFinishedDefaultResponse = Awaited<ReturnType<typeof DagRunService.waitDagRunUntilFinished>>;
export type DagRunServiceWaitDagRunUntilFinishedQueryResult<TData = DagRunServiceWaitDagRunUntilFinishedDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
export const useDagRunServiceWaitDagRunUntilFinishedKey = "DagRunServiceWaitDagRunUntilFinished";
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
index 8bc9e9df8e6..c0ca5feb6cb 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -299,12 +299,14 @@ export const ensureUseDagRunServiceGetUpstreamAssetEventsData = (queryClient: Qu
* @param data.triggeringUserNamePattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
* @param data.dagIdPattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
* @param data.partitionKeyPattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
+* @param data.consumingAssetPattern Filter by consuming asset name or URI using pattern matching
* @returns DAGRunCollectionResponse Successful Response
* @throws ApiError
*/
-export const ensureUseDagRunServiceGetDagRunsData = (queryClient: QueryClient, { bundleVersion, confContains, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, startDateLte, state, triggeringUs [...]
+export const ensureUseDagRunServiceGetDagRunsData = (queryClient: QueryClient, { bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, startDateL [...]
bundleVersion?: string;
confContains?: string;
+ consumingAssetPattern?: string;
dagId: string;
dagIdPattern?: string;
dagVersion?: number[];
@@ -340,7 +342,7 @@ export const ensureUseDagRunServiceGetDagRunsData = (queryClient: QueryClient, {
updatedAtGte?: string;
updatedAtLt?: string;
updatedAtLte?: string;
-}) => queryClient.ensureQueryData({ queryKey: Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, startDateLte, state, trigge [...]
+}) => queryClient.ensureQueryData({ queryKey: Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, star [...]
/**
* Experimental: Wait for a dag run to complete, and return task results if requested.
* 🚧 This is an experimental endpoint and may change or be removed without notice.Successful response are streamed as newline-delimited JSON (NDJSON). Each line is a JSON object representing the DAG run state.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
index f4cb6f482bd..d7c8731e6a6 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -299,12 +299,14 @@ export const prefetchUseDagRunServiceGetUpstreamAssetEvents = (queryClient: Quer
* @param data.triggeringUserNamePattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
* @param data.dagIdPattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
* @param data.partitionKeyPattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
+* @param data.consumingAssetPattern Filter by consuming asset name or URI using pattern matching
* @returns DAGRunCollectionResponse Successful Response
* @throws ApiError
*/
-export const prefetchUseDagRunServiceGetDagRuns = (queryClient: QueryClient, { bundleVersion, confContains, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, startDateLte, state, triggeringUser [...]
+export const prefetchUseDagRunServiceGetDagRuns = (queryClient: QueryClient, { bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, startDateLte [...]
bundleVersion?: string;
confContains?: string;
+ consumingAssetPattern?: string;
dagId: string;
dagIdPattern?: string;
dagVersion?: number[];
@@ -340,7 +342,7 @@ export const prefetchUseDagRunServiceGetDagRuns = (queryClient: QueryClient, { b
updatedAtGte?: string;
updatedAtLt?: string;
updatedAtLte?: string;
-}) => queryClient.prefetchQuery({ queryKey: Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, startDateLte, state, triggeri [...]
+}) => queryClient.prefetchQuery({ queryKey: Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte, startDateLt, startD [...]
/**
* Experimental: Wait for a dag run to complete, and return task results if requested.
* 🚧 This is an experimental endpoint and may change or be removed without notice.Successful response are streamed as newline-delimited JSON (NDJSON). Each line is a JSON object representing the DAG run state.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
index dac7a198e59..019d3ce4a94 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -299,12 +299,14 @@ export const useDagRunServiceGetUpstreamAssetEvents = <TData = Common.DagRunServ
* @param data.triggeringUserNamePattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
* @param data.dagIdPattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
* @param data.partitionKeyPattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
+* @param data.consumingAssetPattern Filter by consuming asset name or URI using pattern matching
* @returns DAGRunCollectionResponse Successful Response
* @throws ApiError
*/
-export const useDagRunServiceGetDagRuns = <TData = Common.DagRunServiceGetDagRunsDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ bundleVersion, confContains, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, [...]
+export const useDagRunServiceGetDagRuns = <TData = Common.DagRunServiceGetDagRunsDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, run [...]
bundleVersion?: string;
confContains?: string;
+ consumingAssetPattern?: string;
dagId: string;
dagIdPattern?: string;
dagVersion?: number[];
@@ -340,7 +342,7 @@ export const useDagRunServiceGetDagRuns = <TData = Common.DagRunServiceGetDagRun
updatedAtGte?: string;
updatedAtLt?: string;
updatedAtLte?: string;
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey: Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, [...]
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey: Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, ru [...]
/**
* Experimental: Wait for a dag run to complete, and return task results if requested.
* 🚧 This is an experimental endpoint and may change or be removed without notice.Successful response are streamed as newline-delimited JSON (NDJSON). Each line is a JSON object representing the DAG run state.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
index c4a41691b1a..cfc50f2d883 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -299,12 +299,14 @@ export const useDagRunServiceGetUpstreamAssetEventsSuspense = <TData = Common.Da
* @param data.triggeringUserNamePattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
* @param data.dagIdPattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
* @param data.partitionKeyPattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
+* @param data.consumingAssetPattern Filter by consuming asset name or URI using pattern matching
* @returns DAGRunCollectionResponse Successful Response
* @throws ApiError
*/
-export const useDagRunServiceGetDagRunsSuspense = <TData = Common.DagRunServiceGetDagRunsDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ bundleVersion, confContains, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runId [...]
+export const useDagRunServiceGetDagRunsSuspense = <TData = Common.DagRunServiceGetDagRunsDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfte [...]
bundleVersion?: string;
confContains?: string;
+ consumingAssetPattern?: string;
dagId: string;
dagIdPattern?: string;
dagVersion?: number[];
@@ -340,7 +342,7 @@ export const useDagRunServiceGetDagRunsSuspense = <TData = Common.DagRunServiceG
updatedAtGte?: string;
updatedAtLt?: string;
updatedAtLte?: string;
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey: Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAf [...]
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey: Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfte [...]
/**
* Experimental: Wait for a dag run to complete, and return task results if requested.
* 🚧 This is an experimental endpoint and may change or be removed without notice.Successful response are streamed as newline-delimited JSON (NDJSON). Each line is a JSON object representing the DAG run state.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
index 6f0c2af82fb..1fbb2db01f5 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -1013,6 +1013,7 @@ export class DagRunService {
* @param data.triggeringUserNamePattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
* @param data.dagIdPattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
* @param data.partitionKeyPattern SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
+ * @param data.consumingAssetPattern Filter by consuming asset name or URI using pattern matching
* @returns DAGRunCollectionResponse Successful Response
* @throws ApiError
*/
@@ -1059,7 +1060,8 @@ export class DagRunService {
run_id_pattern: data.runIdPattern,
triggering_user_name_pattern: data.triggeringUserNamePattern,
dag_id_pattern: data.dagIdPattern,
- partition_key_pattern: data.partitionKeyPattern
+ partition_key_pattern: data.partitionKeyPattern,
+ consuming_asset_pattern: data.consumingAssetPattern
},
errors: {
401: 'Unauthorized',
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 c4dba0647d3..fdb6e663d48 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
@@ -2562,6 +2562,10 @@ export type ClearDagRunResponse = TaskInstanceCollectionResponse | DAGRunRespons
export type GetDagRunsData = {
bundleVersion?: string | null;
confContains?: string;
+ /**
+ * Filter by consuming asset name or URI using pattern matching
+ */
+ consumingAssetPattern?: string | null;
dagId: string;
/**
* SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions are **not** supported.
diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
index 0bbea8fc42d..39bfe2ee791 100644
--- a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
+++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
@@ -28,6 +28,7 @@
},
"collapseAllExtra": "Collapse all extra JSON",
"collapseDetailsPanel": "Collapse Details Panel",
+ "consumingAsset": "Consuming Asset",
"createdAssetEvent_one": "Created Asset Event",
"createdAssetEvent_other": "Created Asset Events",
"dag_one": "Dag",
@@ -131,6 +132,7 @@
"logicalDateTo": "Logical Date To",
"runAfterFrom": "Run After From",
"runAfterTo": "Run After To",
+ "searchAsset": "Search Asset",
"selectDateRange": "Select Date Range",
"startTime": "Start Time"
},
diff --git a/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx b/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx
index 691ed35c042..4b06175db88 100644
--- a/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx
+++ b/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx
@@ -21,7 +21,7 @@
import { Flex } from "@chakra-ui/react";
import { useTranslation } from "react-i18next";
import { BiTargetLock } from "react-icons/bi";
-import { FiBarChart, FiUser } from "react-icons/fi";
+import { FiBarChart, FiUser, FiDatabase } from "react-icons/fi";
import { LuBrackets } from "react-icons/lu";
import {
MdDateRange,
@@ -89,6 +89,13 @@ export const useFilterConfigs = () => {
label: translate("common:dagRun.conf"),
type: FilterTypes.TEXT,
},
+ [SearchParamsKeys.CONSUMING_ASSET_PATTERN]: {
+ hotkeyDisabled: true,
+ icon: <FiDatabase />,
+ label: translate("common:consumingAsset"),
+ placeholder: translate("common:filters.searchAsset"),
+ type: FilterTypes.TEXT,
+ },
[SearchParamsKeys.CREATED_AT_RANGE]: {
endKey: SearchParamsKeys.CREATED_AT_LTE,
icon: <MdDateRange />,
diff --git a/airflow-core/src/airflow/ui/src/constants/searchParams.ts b/airflow-core/src/airflow/ui/src/constants/searchParams.ts
index d37001e7688..7194fe69cee 100644
--- a/airflow-core/src/airflow/ui/src/constants/searchParams.ts
+++ b/airflow-core/src/airflow/ui/src/constants/searchParams.ts
@@ -23,6 +23,7 @@ export enum SearchParamsKeys {
BODY_SEARCH = "body_search",
BUNDLE_VERSION = "bundle_version",
CONF_CONTAINS = "conf_contains",
+ CONSUMING_ASSET_PATTERN = "consuming_asset_pattern",
CREATED_AT_GTE = "created_at_gte",
CREATED_AT_LTE = "created_at_lte",
CREATED_AT_RANGE = "created_at_range",
diff --git a/airflow-core/src/airflow/ui/src/pages/DagRuns.tsx b/airflow-core/src/airflow/ui/src/pages/DagRuns.tsx
index 94b33640312..08eb817db5b 100644
--- a/airflow-core/src/airflow/ui/src/pages/DagRuns.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/DagRuns.tsx
@@ -47,6 +47,7 @@ type DagRunRow = { row: { original: DAGRunResponse } };
const {
BUNDLE_VERSION: BUNDLE_VERSION_PARAM,
CONF_CONTAINS: CONF_CONTAINS_PARAM,
+ CONSUMING_ASSET_PATTERN: CONSUMING_ASSET_PATTERN_PARAM,
DAG_ID_PATTERN: DAG_ID_PATTERN_PARAM,
DAG_VERSION: DAG_VERSION_PARAM,
DURATION_GTE: DURATION_GTE_PARAM,
@@ -214,6 +215,7 @@ export const DagRuns = () => {
const filteredTriggeringUserNamePattern = searchParams.get(TRIGGERING_USER_NAME_PATTERN_PARAM);
const filteredDagIdPattern = searchParams.get(DAG_ID_PATTERN_PARAM);
const filteredDagVersion = searchParams.get(DAG_VERSION_PARAM);
+ const filteredConsumingAsset = searchParams.get(CONSUMING_ASSET_PATTERN_PARAM);
const bundleVersion = searchParams.get(BUNDLE_VERSION_PARAM);
const startDateGte = searchParams.get(START_DATE_GTE_PARAM);
const startDateLte = searchParams.get(START_DATE_LTE_PARAM);
@@ -234,6 +236,7 @@ export const DagRuns = () => {
{
bundleVersion: bundleVersion ?? undefined,
confContains: confContains !== null && confContains !== "" ? confContains : undefined,
+ consumingAssetPattern: filteredConsumingAsset ?? undefined,
dagId: dagId ?? "~",
dagIdPattern: filteredDagIdPattern ?? undefined,
dagVersion:
diff --git a/airflow-core/src/airflow/ui/src/pages/DagRunsFilters.tsx b/airflow-core/src/airflow/ui/src/pages/DagRunsFilters.tsx
index b0f90d66f98..f39e66cab41 100644
--- a/airflow-core/src/airflow/ui/src/pages/DagRunsFilters.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/DagRunsFilters.tsx
@@ -42,6 +42,7 @@ export const DagRunsFilters = ({ dagId }: DagRunsFiltersProps) => {
SearchParamsKeys.DAG_VERSION,
SearchParamsKeys.PARTITION_KEY_PATTERN,
SearchParamsKeys.BUNDLE_VERSION,
+ SearchParamsKeys.CONSUMING_ASSET_PATTERN,
];
if (dagId === undefined) {
diff --git a/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts b/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts
index 17ab2fc3506..6df86f6f6ab 100644
--- a/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts
+++ b/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts
@@ -60,6 +60,7 @@ export type FilterableSearchParamsKeys =
| SearchParamsKeys.BODY_SEARCH
| SearchParamsKeys.BUNDLE_VERSION
| SearchParamsKeys.CONF_CONTAINS
+ | SearchParamsKeys.CONSUMING_ASSET_PATTERN
| SearchParamsKeys.CREATED_AT_RANGE
| SearchParamsKeys.DAG_DISPLAY_NAME_PATTERN
| SearchParamsKeys.DAG_ID
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 2e80a3501fe..9d27392de92 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
@@ -40,6 +40,7 @@ from airflow.utils.types import DagRunTriggeredByType, DagRunType
from tests_common.test_utils.api_fastapi import _check_dag_run_note, _check_last_log
from tests_common.test_utils.asserts import assert_queries_count
from tests_common.test_utils.db import (
+ clear_db_assets,
clear_db_connections,
clear_db_dag_bundles,
clear_db_dags,
@@ -130,6 +131,7 @@ def setup(request, dag_maker, session=None):
clear_db_dag_bundles()
clear_db_serialized_dags()
clear_db_logs()
+ clear_db_assets()
if "no_setup" in request.keywords:
return
@@ -220,6 +222,34 @@ def setup(request, dag_maker, session=None):
session.merge(dag_maker.dag_model)
session.commit()
+ asset1 = AssetModel(name="sales", uri="s3://bucket/sales")
+ asset2 = AssetModel(name="customer", uri="s3://bucket/customer")
+ session.add_all([asset1, asset2])
+ session.flush()
+
+ event1 = AssetEvent(
+ asset_id=asset1.id,
+ source_dag_id="source_dag",
+ source_run_id="source_run",
+ source_task_id="source_task",
+ )
+ event2 = AssetEvent(
+ asset_id=asset2.id,
+ source_dag_id="source_dag",
+ source_run_id="source_run",
+ source_task_id="source_task",
+ )
+ session.add_all([event1, event2])
+ session.flush()
+
+ dag_run1 = session.scalar(select(DagRun).filter(DagRun.id == dag_run1.id))
+ dag_run2 = session.scalar(select(DagRun).filter(DagRun.id == dag_run2.id))
+
+ dag_run1.consumed_asset_events.append(event1)
+ dag_run2.consumed_asset_events.append(event2)
+
+ session.commit()
+
def get_dag_versions_dict(dag_versions: list[DagVersion]) -> list[dict]:
return [
@@ -710,6 +740,21 @@ class TestGetDagRuns:
), # Test for debug key
("~", {"conf_contains": "version"}, [DAG1_RUN1_ID]), # Test for the key "version"
("~", {"conf_contains": "nonexistent_key"}, []), # Test for a key that doesn't exist
+ # Test consuming_asset_pattern filter
+ ("~", {"consuming_asset_pattern": "sales"}, [DAG1_RUN1_ID]), # Filter by asset name
+ ("~", {"consuming_asset_pattern": "s3://bucket/sales"}, [DAG1_RUN1_ID]), # Filter by asset URI
+ ("~", {"consuming_asset_pattern": "customer"}, [DAG1_RUN2_ID]), # Filter by another asset
+ (
+ "~",
+ {"consuming_asset_pattern": "s3://bucket/customer"},
+ [DAG1_RUN2_ID],
+ ), # Filter by customer URI
+ (
+ "~",
+ {"consuming_asset_pattern": "s3://bucket"},
+ [DAG1_RUN1_ID, DAG1_RUN2_ID],
+ ), # Partial URI match
+ ("~", {"consuming_asset_pattern": "nonexistent_asset"}, []), # Non-existent asset returns empty
],
)
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")