This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun 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 4cb967fd5c2 Add /ui/assets endpoint with last-event sort and use it in
the Assets list (#69940)
4cb967fd5c2 is described below
commit 4cb967fd5c2304e65d05721100d6d59d2b27b266
Author: Brent Bovenzi <[email protected]>
AuthorDate: Thu Jul 30 04:32:23 2026 -0400
Add /ui/assets endpoint with last-event sort and use it in the Assets list
(#69940)
The Assets list needs recency sorting and richer filtering (by group and
last
event time) that the public /assets endpoint intentionally does not expose.
Add a dedicated UI-only endpoint for those needs and point the Assets list
at
it, leaving the public API contract untouched.
---
.../src/airflow/api_fastapi/common/db/assets.py | 54 +++
.../src/airflow/api_fastapi/common/parameters.py | 6 +
.../api_fastapi/core_api/datamodels/assets.py | 29 ++
.../api_fastapi/core_api/openapi/_private_ui.yaml | 415 +++++++++++++++++++++
.../api_fastapi/core_api/routes/ui/assets.py | 97 ++++-
.../src/airflow/ui/openapi-gen/queries/common.ts | 21 ++
.../ui/openapi-gen/queries/ensureQueryData.ts | 41 ++
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 41 ++
.../src/airflow/ui/openapi-gen/queries/queries.ts | 41 ++
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 41 ++
.../ui/openapi-gen/requests/services.gen.ts | 53 ++-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 60 +++
.../airflow/ui/public/i18n/locales/en/assets.json | 5 +-
.../src/airflow/ui/src/constants/filterConfigs.tsx | 17 +-
.../src/airflow/ui/src/constants/searchParams.ts | 4 +
.../airflow/ui/src/pages/AssetsList/AssetsList.tsx | 84 ++---
.../src/airflow/ui/src/utils/useFiltersHandler.ts | 2 +
.../api_fastapi/core_api/routes/ui/test_assets.py | 286 +++++++++++++-
18 files changed, 1245 insertions(+), 52 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
new file mode 100644
index 00000000000..e393ea19060
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/common/db/assets.py
@@ -0,0 +1,54 @@
+# 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.
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import subqueryload
+
+from airflow.models.asset import AssetEvent, AssetModel, AssetWatcherModel
+
+if TYPE_CHECKING:
+ from sqlalchemy.sql import Select
+
+
+def generate_assets_with_last_event_query() -> Select:
+ """Fetch Assets outer-joined to their latest AssetEvent id/timestamp."""
+ max_asset_event_id_query = (
+ select(AssetEvent.asset_id,
func.max(AssetEvent.id).label("max_asset_event_id"))
+ .group_by(AssetEvent.asset_id)
+ .subquery()
+ )
+
+ return (
+ select(
+ AssetModel,
+ AssetEvent.id.label("last_asset_event_id"),
+ AssetEvent.timestamp.label("last_asset_event_timestamp"),
+ )
+ .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(
+ subqueryload(AssetModel.scheduled_dags),
+ subqueryload(AssetModel.producing_tasks),
+ subqueryload(AssetModel.consuming_tasks),
+ subqueryload(AssetModel.aliases),
+
subqueryload(AssetModel.watchers).joinedload(AssetWatcherModel.trigger),
+ )
+ )
diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters.py
b/airflow-core/src/airflow/api_fastapi/common/parameters.py
index 1284fdacb66..76fe643fd6b 100644
--- a/airflow-core/src/airflow/api_fastapi/common/parameters.py
+++ b/airflow-core/src/airflow/api_fastapi/common/parameters.py
@@ -1812,6 +1812,12 @@ QueryUriExactMatch = Annotated[
)
),
]
+QueryAssetGroupPatternSearch = Annotated[
+ _SearchParam, Depends(search_param_factory(AssetModel.group,
"group_pattern"))
+]
+QueryAssetGroupPrefixPatternSearch = Annotated[
+ _PrefixSearchParam, Depends(prefix_search_param_factory(AssetModel.group,
"group_prefix_pattern"))
+]
QueryAssetAliasNamePatternSearch = Annotated[
_SearchParam, Depends(search_param_factory(AssetAliasModel.name,
"name_pattern"))
]
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 3be0707bee1..8966891cf8b 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
@@ -40,6 +40,7 @@ from airflow.models.base import ID_LEN
from airflow.utils.types import DagRunType
if TYPE_CHECKING:
+ from airflow.models.asset import AssetModel
from airflow.serialization.definitions.dag import SerializedDAG
@@ -106,6 +107,34 @@ class AssetResponse(BaseModel):
def redact_extra(cls, v: dict):
return redact(v)
+ @classmethod
+ def from_asset_row(
+ cls,
+ asset: AssetModel,
+ last_asset_event_id: int | None,
+ last_asset_event_timestamp: datetime | None,
+ ) -> AssetResponse:
+ """Build a response from an ``AssetModel`` row joined with its last
AssetEvent id/timestamp."""
+ watchers_data = [
+ {
+ "name": watcher.name,
+ "trigger_id": watcher.trigger_id,
+ "created_date": watcher.trigger.created_date,
+ }
+ for watcher in asset.watchers
+ ]
+ return cls.model_validate(
+ {
+ **asset.__dict__,
+ "aliases": asset.aliases,
+ "watchers": watchers_data,
+ "last_asset_event": {
+ "id": last_asset_event_id,
+ "timestamp": last_asset_event_timestamp,
+ },
+ }
+ )
+
class AssetCollectionResponse(BaseModel):
"""Asset collection response."""
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 2237e622d75..a6d31b9956b 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
@@ -81,6 +81,206 @@ paths:
security:
- OAuth2PasswordBearer: []
- HTTPBearer: []
+ /ui/assets:
+ get:
+ tags:
+ - Asset
+ summary: Get Assets
+ description: Get assets. Like the public endpoint, but also supports
sorting
+ by group and last asset event timestamp.
+ operationId: get_assets_ui
+ security:
+ - OAuth2PasswordBearer: []
+ - HTTPBearer: []
+ parameters:
+ - name: limit
+ in: query
+ required: false
+ schema:
+ type: integer
+ minimum: 0
+ default: 50
+ title: Limit
+ - name: offset
+ in: query
+ required: false
+ schema:
+ type: integer
+ minimum: 0
+ default: 0
+ title: Offset
+ - name: name_pattern
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ description: "Case-insensitive substring match (SQL `ILIKE`). Slower
than\
+ \ `name_prefix_pattern` on large tables \u2014 see \"Filtering
with pattern\
+ \ parameters\"."
+ title: Name Pattern
+ description: "Case-insensitive substring match (SQL `ILIKE`). Slower
than\
+ \ `name_prefix_pattern` on large tables \u2014 see \"Filtering with
pattern\
+ \ parameters\"."
+ - name: name_prefix_pattern
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ description: Case-sensitive, index-friendly prefix match. See
"Filtering
+ with pattern parameters".
+ title: Name Prefix Pattern
+ description: Case-sensitive, index-friendly prefix match. See
"Filtering with
+ pattern parameters".
+ - name: uri
+ in: query
+ required: false
+ schema:
+ type: array
+ items:
+ type: string
+ description: Exact-match filter on the full asset URI. Compiles to
an indexed
+ equality comparison (``uri = ...``). Repeat the parameter
(``?uri=a&uri=b``)
+ to match multiple assets.
+ title: Uri
+ description: Exact-match filter on the full asset URI. Compiles to an
indexed
+ equality comparison (``uri = ...``). Repeat the parameter
(``?uri=a&uri=b``)
+ to match multiple assets.
+ - name: uri_pattern
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ description: "Case-insensitive substring match (SQL `ILIKE`). Slower
than\
+ \ `uri_prefix_pattern` on large tables \u2014 see \"Filtering with
pattern\
+ \ parameters\"."
+ title: Uri Pattern
+ description: "Case-insensitive substring match (SQL `ILIKE`). Slower
than\
+ \ `uri_prefix_pattern` on large tables \u2014 see \"Filtering with
pattern\
+ \ parameters\"."
+ - name: uri_prefix_pattern
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ description: Case-sensitive, index-friendly prefix match. See
"Filtering
+ with pattern parameters".
+ title: Uri Prefix Pattern
+ description: Case-sensitive, index-friendly prefix match. See
"Filtering with
+ pattern parameters".
+ - name: group_pattern
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ description: "Case-insensitive substring match (SQL `ILIKE`). Slower
than\
+ \ `group_prefix_pattern` on large tables \u2014 see \"Filtering
with pattern\
+ \ parameters\"."
+ title: Group Pattern
+ description: "Case-insensitive substring match (SQL `ILIKE`). Slower
than\
+ \ `group_prefix_pattern` on large tables \u2014 see \"Filtering with
pattern\
+ \ parameters\"."
+ - name: group_prefix_pattern
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ description: Case-sensitive, index-friendly prefix match. See
"Filtering
+ with pattern parameters".
+ title: Group Prefix Pattern
+ description: Case-sensitive, index-friendly prefix match. See
"Filtering with
+ pattern parameters".
+ - name: dag_ids
+ in: query
+ required: false
+ schema:
+ type: array
+ items:
+ type: string
+ title: Dag Ids
+ - name: only_active
+ in: query
+ required: false
+ schema:
+ type: boolean
+ default: true
+ title: Only Active
+ - name: last_asset_event_timestamp_gte
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Last Asset Event Timestamp Gte
+ - name: last_asset_event_timestamp_gt
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Last Asset Event Timestamp Gt
+ - name: last_asset_event_timestamp_lte
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Last Asset Event Timestamp Lte
+ - name: last_asset_event_timestamp_lt
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Last Asset Event Timestamp Lt
+ - name: order_by
+ in: query
+ required: false
+ schema:
+ type: array
+ items:
+ type: string
+ description: 'Attributes to order by, multi criteria sort is
supported.
+ Prefix with `-` for descending order. Supported attributes: `id,
name,
+ uri, group, created_at, updated_at, last_asset_event_timestamp`'
+ default:
+ - -last_asset_event_timestamp
+ title: Order By
+ description: 'Attributes to order by, multi criteria sort is
supported. Prefix
+ with `-` for descending order. Supported attributes: `id, name, uri,
group,
+ created_at, updated_at, last_asset_event_timestamp`'
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/AssetCollectionResponse'
+ '422':
+ description: Validation Error
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPValidationError'
/ui/next_run_assets/{dag_id}:
get:
tags:
@@ -1918,6 +2118,40 @@ paths:
$ref: '#/components/schemas/HTTPValidationError'
components:
schemas:
+ AssetAliasResponse:
+ properties:
+ id:
+ type: integer
+ title: Id
+ name:
+ type: string
+ title: Name
+ group:
+ type: string
+ title: Group
+ type: object
+ required:
+ - id
+ - name
+ - group
+ title: AssetAliasResponse
+ description: Asset alias serializer for responses.
+ AssetCollectionResponse:
+ properties:
+ assets:
+ items:
+ $ref: '#/components/schemas/AssetResponse'
+ type: array
+ title: Assets
+ total_entries:
+ type: integer
+ title: Total Entries
+ type: object
+ required:
+ - assets
+ - total_entries
+ title: AssetCollectionResponse
+ description: Asset collection response.
AssetExpressionAlias:
properties:
alias:
@@ -2035,6 +2269,98 @@ components:
title: AssetExpressionRef
description: 'An unresolved asset reference leaf: ``{"asset_ref":
{"name": ...}}``
or ``{"asset_ref": {"uri": ...}}``.'
+ AssetResponse:
+ properties:
+ id:
+ type: integer
+ title: Id
+ name:
+ type: string
+ title: Name
+ uri:
+ type: string
+ title: Uri
+ group:
+ type: string
+ title: Group
+ extra:
+ anyOf:
+ - additionalProperties:
+ $ref: '#/components/schemas/JsonValue'
+ type: object
+ - type: 'null'
+ title: Extra
+ created_at:
+ type: string
+ format: date-time
+ title: Created At
+ updated_at:
+ type: string
+ format: date-time
+ title: Updated At
+ scheduled_dags:
+ items:
+ $ref: '#/components/schemas/DagScheduleAssetReference'
+ type: array
+ title: Scheduled Dags
+ producing_tasks:
+ items:
+ $ref: '#/components/schemas/TaskOutletAssetReference'
+ type: array
+ title: Producing Tasks
+ consuming_tasks:
+ items:
+ $ref: '#/components/schemas/TaskInletAssetReference'
+ type: array
+ title: Consuming Tasks
+ aliases:
+ items:
+ $ref: '#/components/schemas/AssetAliasResponse'
+ type: array
+ title: Aliases
+ watchers:
+ items:
+ $ref: '#/components/schemas/AssetWatcherResponse'
+ type: array
+ title: Watchers
+ last_asset_event:
+ anyOf:
+ - $ref: '#/components/schemas/LastAssetEventResponse'
+ - type: 'null'
+ type: object
+ required:
+ - id
+ - name
+ - uri
+ - group
+ - created_at
+ - updated_at
+ - scheduled_dags
+ - producing_tasks
+ - consuming_tasks
+ - aliases
+ - watchers
+ title: AssetResponse
+ description: Asset serializer for responses.
+ AssetWatcherResponse:
+ properties:
+ name:
+ type: string
+ title: Name
+ trigger_id:
+ type: integer
+ title: Trigger Id
+ created_date:
+ type: string
+ format: date-time
+ title: Created Date
+ type: object
+ required:
+ - name
+ - trigger_id
+ - created_date
+ title: AssetWatcherResponse
+ description: Asset watcher serializer for responses.
AuthenticatedMeResponse:
properties:
id:
@@ -2786,6 +3112,27 @@ components:
- asset_materialization
title: DagRunType
description: Class with DagRun types.
+ DagScheduleAssetReference:
+ properties:
+ dag_id:
+ type: string
+ title: Dag Id
+ created_at:
+ type: string
+ format: date-time
+ title: Created At
+ updated_at:
+ type: string
+ format: date-time
+ title: Updated At
+ additionalProperties: false
+ type: object
+ required:
+ - dag_id
+ - created_at
+ - updated_at
+ title: DagScheduleAssetReference
+ description: Dag schedule reference serializer for assets.
DagTagResponse:
properties:
name:
@@ -3498,6 +3845,24 @@ components:
- unixname
title: JobResponse
description: Job serializer for responses.
+ JsonValue: {}
+ LastAssetEventResponse:
+ properties:
+ id:
+ anyOf:
+ - type: integer
+ minimum: 0.0
+ - type: 'null'
+ title: Id
+ timestamp:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Timestamp
+ type: object
+ title: LastAssetEventResponse
+ description: Last asset event response serializer.
LightGridTaskInstanceSummary:
properties:
task_id:
@@ -4000,6 +4365,31 @@ components:
- nodes
title: StructureDataResponse
description: Structure Data serializer for responses.
+ TaskInletAssetReference:
+ properties:
+ dag_id:
+ type: string
+ title: Dag Id
+ task_id:
+ type: string
+ title: Task Id
+ created_at:
+ type: string
+ format: date-time
+ title: Created At
+ updated_at:
+ type: string
+ format: date-time
+ title: Updated At
+ additionalProperties: false
+ type: object
+ required:
+ - dag_id
+ - task_id
+ - created_at
+ - updated_at
+ title: TaskInletAssetReference
+ description: Task inlet reference serializer for assets.
TaskInstanceResponse:
properties:
id:
@@ -4273,6 +4663,31 @@ components:
- awaiting_input
title: TaskInstanceStateCount
description: TaskInstance serializer for responses.
+ TaskOutletAssetReference:
+ properties:
+ dag_id:
+ type: string
+ title: Dag Id
+ task_id:
+ type: string
+ title: Task Id
+ created_at:
+ type: string
+ format: date-time
+ title: Created At
+ updated_at:
+ type: string
+ format: date-time
+ title: Updated At
+ additionalProperties: false
+ type: object
+ required:
+ - dag_id
+ - task_id
+ - created_at
+ - updated_at
+ title: TaskOutletAssetReference
+ description: Task outlet reference serializer for assets.
TeamCollectionResponse:
properties:
teams:
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py
index d3da95178a1..61384896625 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py
@@ -16,20 +16,42 @@
# under the License.
from __future__ import annotations
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Annotated, Any, cast
import structlog
from fastapi import Depends, HTTPException, status
from sqlalchemy import ColumnElement, and_, case, exists, func, select, true
-from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.common.db.assets import
generate_assets_with_last_event_query
+from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.parameters import (
+ QueryAssetDagIdPatternSearch,
+ QueryAssetGroupPatternSearch,
+ QueryAssetGroupPrefixPatternSearch,
+ QueryAssetNamePatternSearch,
+ QueryAssetNamePrefixPatternSearch,
+ QueryLimit,
+ QueryOffset,
+ QueryUriExactMatch,
+ QueryUriPatternSearch,
+ QueryUriPrefixPatternSearch,
+ RangeFilter,
+ SortParam,
+ datetime_range_filter_factory,
+)
from airflow.api_fastapi.common.partition_helpers import
load_partitioned_timetable
from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.assets import
AssetCollectionResponse, AssetResponse
from airflow.api_fastapi.core_api.datamodels.ui.assets import (
NextRunAssetEventResponse,
NextRunAssetsResponse,
)
-from airflow.api_fastapi.core_api.security import requires_access_asset,
requires_access_dag
+from airflow.api_fastapi.core_api.routes.public.assets import OnlyActiveFilter
+from airflow.api_fastapi.core_api.security import (
+ requires_access_asset,
+ requires_access_asset_alias,
+ requires_access_dag,
+)
from airflow.models import DagModel
from airflow.models.asset import (
AssetActive,
@@ -49,6 +71,75 @@ log = structlog.get_logger(logger_name=__name__)
assets_router = AirflowRouter(tags=["Asset"])
+@assets_router.get(
+ "/assets",
+ dependencies=[
+ Depends(requires_access_asset(method="GET")),
+ Depends(requires_access_asset_alias(method="GET")),
+ ],
+ operation_id="get_assets_ui",
+)
+def get_assets(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ name_pattern: QueryAssetNamePatternSearch,
+ name_prefix_pattern: QueryAssetNamePrefixPatternSearch,
+ uri: QueryUriExactMatch,
+ uri_pattern: QueryUriPatternSearch,
+ uri_prefix_pattern: QueryUriPrefixPatternSearch,
+ group_pattern: QueryAssetGroupPatternSearch,
+ group_prefix_pattern: QueryAssetGroupPrefixPatternSearch,
+ dag_ids: QueryAssetDagIdPatternSearch,
+ only_active: Annotated[OnlyActiveFilter,
Depends(OnlyActiveFilter.depends)],
+ last_asset_event_timestamp_range: Annotated[
+ RangeFilter,
+ Depends(
+ datetime_range_filter_factory(
+ "last_asset_event_timestamp", AssetEvent,
attribute_name="timestamp"
+ )
+ ),
+ ],
+ order_by: Annotated[
+ SortParam,
+ Depends(
+ SortParam(
+ ["id", "name", "uri", "group", "created_at", "updated_at"],
+ AssetModel,
+ {"last_asset_event_timestamp": AssetEvent.timestamp},
+ ).dynamic_depends(default="-last_asset_event_timestamp")
+ ),
+ ],
+ session: SessionDep,
+) -> AssetCollectionResponse:
+ """Get assets. Like the public endpoint, but also supports sorting by
group and last asset event timestamp."""
+ assets_select, total_entries = paginated_select(
+ statement=generate_assets_with_last_event_query(),
+ filters=[
+ only_active,
+ name_pattern,
+ name_prefix_pattern,
+ uri,
+ uri_pattern,
+ uri_prefix_pattern,
+ group_pattern,
+ group_prefix_pattern,
+ dag_ids,
+ last_asset_event_timestamp_range,
+ ],
+ order_by=order_by,
+ offset=offset,
+ limit=limit,
+ session=session,
+ )
+
+ assets = [
+ AssetResponse.from_asset_row(asset, last_asset_event_id,
last_asset_event_timestamp)
+ for asset, last_asset_event_id, last_asset_event_timestamp in
session.execute(assets_select)
+ ]
+
+ return AssetCollectionResponse(assets=assets, total_entries=total_entries)
+
+
@assets_router.get(
"/next_run_assets/{dag_id}",
dependencies=[Depends(requires_access_asset(method="GET")),
Depends(requires_access_dag(method="GET"))],
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 64e87bd2a1d..0b1fda69bba 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -84,6 +84,27 @@ export const UseAssetServiceGetDagAssetQueuedEventKeyFn = ({
assetId, before, da
before?: string;
dagId: string;
}, queryKey?: Array<unknown>) => [useAssetServiceGetDagAssetQueuedEventKey,
...(queryKey ?? [{ assetId, before, dagId }])];
+export type AssetServiceGetAssetsUiDefaultResponse = Awaited<ReturnType<typeof
AssetService.getAssetsUi>>;
+export type AssetServiceGetAssetsUiQueryResult<TData =
AssetServiceGetAssetsUiDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
+export const useAssetServiceGetAssetsUiKey = "AssetServiceGetAssetsUi";
+export const UseAssetServiceGetAssetsUiKeyFn = ({ dagIds, groupPattern,
groupPrefixPattern, lastAssetEventTimestampGt, lastAssetEventTimestampGte,
lastAssetEventTimestampLt, lastAssetEventTimestampLte, limit, namePattern,
namePrefixPattern, offset, onlyActive, orderBy, uri, uriPattern,
uriPrefixPattern }: {
+ dagIds?: string[];
+ groupPattern?: string;
+ groupPrefixPattern?: string;
+ lastAssetEventTimestampGt?: string;
+ lastAssetEventTimestampGte?: string;
+ lastAssetEventTimestampLt?: string;
+ lastAssetEventTimestampLte?: string;
+ limit?: number;
+ namePattern?: string;
+ namePrefixPattern?: string;
+ offset?: number;
+ onlyActive?: boolean;
+ orderBy?: string[];
+ uri?: string[];
+ uriPattern?: string;
+ uriPrefixPattern?: string;
+} = {}, queryKey?: Array<unknown>) => [useAssetServiceGetAssetsUiKey,
...(queryKey ?? [{ dagIds, groupPattern, groupPrefixPattern,
lastAssetEventTimestampGt, lastAssetEventTimestampGte,
lastAssetEventTimestampLt, lastAssetEventTimestampLte, limit, namePattern,
namePrefixPattern, offset, onlyActive, orderBy, uri, uriPattern,
uriPrefixPattern }])];
export type AssetServiceNextRunAssetsDefaultResponse =
Awaited<ReturnType<typeof AssetService.nextRunAssets>>;
export type AssetServiceNextRunAssetsQueryResult<TData =
AssetServiceNextRunAssetsDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useAssetServiceNextRunAssetsKey = "AssetServiceNextRunAssets";
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 c3c73d7d427..091573243df 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -159,6 +159,47 @@ export const
ensureUseAssetServiceGetDagAssetQueuedEventData = (queryClient: Que
dagId: string;
}) => queryClient.ensureQueryData({ queryKey:
Common.UseAssetServiceGetDagAssetQueuedEventKeyFn({ assetId, before, dagId }),
queryFn: () => AssetService.getDagAssetQueuedEvent({ assetId, before, dagId })
});
/**
+* Get Assets
+* Get assets. Like the public endpoint, but also supports sorting by group and
last asset event timestamp.
+* @param data The data for the request.
+* @param data.limit
+* @param data.offset
+* @param data.namePattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `name_prefix_pattern` on large tables — see "Filtering with pattern
parameters".
+* @param data.namePrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.uri Exact-match filter on the full asset URI. Compiles to an
indexed equality comparison (``uri = ...``). Repeat the parameter
(``?uri=a&uri=b``) to match multiple assets.
+* @param data.uriPattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `uri_prefix_pattern` on large tables — see "Filtering with pattern
parameters".
+* @param data.uriPrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.groupPattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `group_prefix_pattern` on large tables — see "Filtering with
pattern parameters".
+* @param data.groupPrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.dagIds
+* @param data.onlyActive
+* @param data.lastAssetEventTimestampGte
+* @param data.lastAssetEventTimestampGt
+* @param data.lastAssetEventTimestampLte
+* @param data.lastAssetEventTimestampLt
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
name, uri, group, created_at, updated_at, last_asset_event_timestamp`
+* @returns AssetCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const ensureUseAssetServiceGetAssetsUiData = (queryClient: QueryClient,
{ dagIds, groupPattern, groupPrefixPattern, lastAssetEventTimestampGt,
lastAssetEventTimestampGte, lastAssetEventTimestampLt,
lastAssetEventTimestampLte, limit, namePattern, namePrefixPattern, offset,
onlyActive, orderBy, uri, uriPattern, uriPrefixPattern }: {
+ dagIds?: string[];
+ groupPattern?: string;
+ groupPrefixPattern?: string;
+ lastAssetEventTimestampGt?: string;
+ lastAssetEventTimestampGte?: string;
+ lastAssetEventTimestampLt?: string;
+ lastAssetEventTimestampLte?: string;
+ limit?: number;
+ namePattern?: string;
+ namePrefixPattern?: string;
+ offset?: number;
+ onlyActive?: boolean;
+ orderBy?: string[];
+ uri?: string[];
+ uriPattern?: string;
+ uriPrefixPattern?: string;
+} = {}) => queryClient.ensureQueryData({ queryKey:
Common.UseAssetServiceGetAssetsUiKeyFn({ dagIds, groupPattern,
groupPrefixPattern, lastAssetEventTimestampGt, lastAssetEventTimestampGte,
lastAssetEventTimestampLt, lastAssetEventTimestampLte, limit, namePattern,
namePrefixPattern, offset, onlyActive, orderBy, uri, uriPattern,
uriPrefixPattern }), queryFn: () => AssetService.getAssetsUi({ dagIds,
groupPattern, groupPrefixPattern, lastAssetEventTimestampGt,
lastAssetEventTimestampGte, las [...]
+/**
* Next Run Assets
* @param data The data for the request.
* @param data.dagId
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 53e93745026..2550f816011 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -159,6 +159,47 @@ export const prefetchUseAssetServiceGetDagAssetQueuedEvent
= (queryClient: Query
dagId: string;
}) => queryClient.prefetchQuery({ queryKey:
Common.UseAssetServiceGetDagAssetQueuedEventKeyFn({ assetId, before, dagId }),
queryFn: () => AssetService.getDagAssetQueuedEvent({ assetId, before, dagId })
});
/**
+* Get Assets
+* Get assets. Like the public endpoint, but also supports sorting by group and
last asset event timestamp.
+* @param data The data for the request.
+* @param data.limit
+* @param data.offset
+* @param data.namePattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `name_prefix_pattern` on large tables — see "Filtering with pattern
parameters".
+* @param data.namePrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.uri Exact-match filter on the full asset URI. Compiles to an
indexed equality comparison (``uri = ...``). Repeat the parameter
(``?uri=a&uri=b``) to match multiple assets.
+* @param data.uriPattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `uri_prefix_pattern` on large tables — see "Filtering with pattern
parameters".
+* @param data.uriPrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.groupPattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `group_prefix_pattern` on large tables — see "Filtering with
pattern parameters".
+* @param data.groupPrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.dagIds
+* @param data.onlyActive
+* @param data.lastAssetEventTimestampGte
+* @param data.lastAssetEventTimestampGt
+* @param data.lastAssetEventTimestampLte
+* @param data.lastAssetEventTimestampLt
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
name, uri, group, created_at, updated_at, last_asset_event_timestamp`
+* @returns AssetCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const prefetchUseAssetServiceGetAssetsUi = (queryClient: QueryClient, {
dagIds, groupPattern, groupPrefixPattern, lastAssetEventTimestampGt,
lastAssetEventTimestampGte, lastAssetEventTimestampLt,
lastAssetEventTimestampLte, limit, namePattern, namePrefixPattern, offset,
onlyActive, orderBy, uri, uriPattern, uriPrefixPattern }: {
+ dagIds?: string[];
+ groupPattern?: string;
+ groupPrefixPattern?: string;
+ lastAssetEventTimestampGt?: string;
+ lastAssetEventTimestampGte?: string;
+ lastAssetEventTimestampLt?: string;
+ lastAssetEventTimestampLte?: string;
+ limit?: number;
+ namePattern?: string;
+ namePrefixPattern?: string;
+ offset?: number;
+ onlyActive?: boolean;
+ orderBy?: string[];
+ uri?: string[];
+ uriPattern?: string;
+ uriPrefixPattern?: string;
+} = {}) => queryClient.prefetchQuery({ queryKey:
Common.UseAssetServiceGetAssetsUiKeyFn({ dagIds, groupPattern,
groupPrefixPattern, lastAssetEventTimestampGt, lastAssetEventTimestampGte,
lastAssetEventTimestampLt, lastAssetEventTimestampLte, limit, namePattern,
namePrefixPattern, offset, onlyActive, orderBy, uri, uriPattern,
uriPrefixPattern }), queryFn: () => AssetService.getAssetsUi({ dagIds,
groupPattern, groupPrefixPattern, lastAssetEventTimestampGt,
lastAssetEventTimestampGte, lastA [...]
+/**
* Next Run Assets
* @param data The data for the request.
* @param data.dagId
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 49f1fa6d734..9da0d1e5ab1 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -159,6 +159,47 @@ export const useAssetServiceGetDagAssetQueuedEvent =
<TData = Common.AssetServic
dagId: string;
}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseAssetServiceGetDagAssetQueuedEventKeyFn({ assetId, before, dagId },
queryKey), queryFn: () => AssetService.getDagAssetQueuedEvent({ assetId,
before, dagId }) as TData, ...options });
/**
+* Get Assets
+* Get assets. Like the public endpoint, but also supports sorting by group and
last asset event timestamp.
+* @param data The data for the request.
+* @param data.limit
+* @param data.offset
+* @param data.namePattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `name_prefix_pattern` on large tables — see "Filtering with pattern
parameters".
+* @param data.namePrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.uri Exact-match filter on the full asset URI. Compiles to an
indexed equality comparison (``uri = ...``). Repeat the parameter
(``?uri=a&uri=b``) to match multiple assets.
+* @param data.uriPattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `uri_prefix_pattern` on large tables — see "Filtering with pattern
parameters".
+* @param data.uriPrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.groupPattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `group_prefix_pattern` on large tables — see "Filtering with
pattern parameters".
+* @param data.groupPrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.dagIds
+* @param data.onlyActive
+* @param data.lastAssetEventTimestampGte
+* @param data.lastAssetEventTimestampGt
+* @param data.lastAssetEventTimestampLte
+* @param data.lastAssetEventTimestampLt
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
name, uri, group, created_at, updated_at, last_asset_event_timestamp`
+* @returns AssetCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useAssetServiceGetAssetsUi = <TData =
Common.AssetServiceGetAssetsUiDefaultResponse, TError = unknown, TQueryKey
extends Array<unknown> = unknown[]>({ dagIds, groupPattern, groupPrefixPattern,
lastAssetEventTimestampGt, lastAssetEventTimestampGte,
lastAssetEventTimestampLt, lastAssetEventTimestampLte, limit, namePattern,
namePrefixPattern, offset, onlyActive, orderBy, uri, uriPattern,
uriPrefixPattern }: {
+ dagIds?: string[];
+ groupPattern?: string;
+ groupPrefixPattern?: string;
+ lastAssetEventTimestampGt?: string;
+ lastAssetEventTimestampGte?: string;
+ lastAssetEventTimestampLt?: string;
+ lastAssetEventTimestampLte?: string;
+ limit?: number;
+ namePattern?: string;
+ namePrefixPattern?: string;
+ offset?: number;
+ onlyActive?: boolean;
+ orderBy?: string[];
+ uri?: string[];
+ uriPattern?: string;
+ uriPrefixPattern?: string;
+} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseAssetServiceGetAssetsUiKeyFn({ dagIds, groupPattern,
groupPrefixPattern, lastAssetEventTimestampGt, lastAssetEventTimestampGte,
lastAssetEventTimestampLt, lastAssetEventTimestampLte, limit, namePattern,
namePrefixPattern, offset, onlyActive, orderBy, uri, uriPattern,
uriPrefixPattern }, queryKey), queryFn: () => AssetService.getAssetsUi({ [...]
+/**
* Next Run Assets
* @param data The data for the request.
* @param data.dagId
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 a67ccaa19ee..2b694fe3b5a 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -159,6 +159,47 @@ export const useAssetServiceGetDagAssetQueuedEventSuspense
= <TData = Common.Ass
dagId: string;
}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseAssetServiceGetDagAssetQueuedEventKeyFn({ assetId, before, dagId },
queryKey), queryFn: () => AssetService.getDagAssetQueuedEvent({ assetId,
before, dagId }) as TData, ...options });
/**
+* Get Assets
+* Get assets. Like the public endpoint, but also supports sorting by group and
last asset event timestamp.
+* @param data The data for the request.
+* @param data.limit
+* @param data.offset
+* @param data.namePattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `name_prefix_pattern` on large tables — see "Filtering with pattern
parameters".
+* @param data.namePrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.uri Exact-match filter on the full asset URI. Compiles to an
indexed equality comparison (``uri = ...``). Repeat the parameter
(``?uri=a&uri=b``) to match multiple assets.
+* @param data.uriPattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `uri_prefix_pattern` on large tables — see "Filtering with pattern
parameters".
+* @param data.uriPrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.groupPattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `group_prefix_pattern` on large tables — see "Filtering with
pattern parameters".
+* @param data.groupPrefixPattern Case-sensitive, index-friendly prefix match.
See "Filtering with pattern parameters".
+* @param data.dagIds
+* @param data.onlyActive
+* @param data.lastAssetEventTimestampGte
+* @param data.lastAssetEventTimestampGt
+* @param data.lastAssetEventTimestampLte
+* @param data.lastAssetEventTimestampLt
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
name, uri, group, created_at, updated_at, last_asset_event_timestamp`
+* @returns AssetCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useAssetServiceGetAssetsUiSuspense = <TData =
Common.AssetServiceGetAssetsUiDefaultResponse, TError = unknown, TQueryKey
extends Array<unknown> = unknown[]>({ dagIds, groupPattern, groupPrefixPattern,
lastAssetEventTimestampGt, lastAssetEventTimestampGte,
lastAssetEventTimestampLt, lastAssetEventTimestampLte, limit, namePattern,
namePrefixPattern, offset, onlyActive, orderBy, uri, uriPattern,
uriPrefixPattern }: {
+ dagIds?: string[];
+ groupPattern?: string;
+ groupPrefixPattern?: string;
+ lastAssetEventTimestampGt?: string;
+ lastAssetEventTimestampGte?: string;
+ lastAssetEventTimestampLt?: string;
+ lastAssetEventTimestampLte?: string;
+ limit?: number;
+ namePattern?: string;
+ namePrefixPattern?: string;
+ offset?: number;
+ onlyActive?: boolean;
+ orderBy?: string[];
+ uri?: string[];
+ uriPattern?: string;
+ uriPrefixPattern?: string;
+} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseAssetServiceGetAssetsUiKeyFn({ dagIds, groupPattern,
groupPrefixPattern, lastAssetEventTimestampGt, lastAssetEventTimestampGte,
lastAssetEventTimestampLt, lastAssetEventTimestampLte, limit, namePattern,
namePrefixPattern, offset, onlyActive, orderBy, uri, uriPattern,
uriPrefixPattern }, queryKey), queryFn: () => AssetService.getAss [...]
+/**
* Next Run Assets
* @param data The data for the request.
* @param data.dagId
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 d34910b8276..c31f14f957b 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
@@ -3,7 +3,7 @@
import type { CancelablePromise } from './core/CancelablePromise';
import { OpenAPI } from './core/OpenAPI';
import { request as __request } from './core/request';
-import type { GetAssetsData, GetAssetsResponse, GetAssetAliasesData,
GetAssetAliasesResponse, GetAssetAliasData, GetAssetAliasResponse,
GetAssetEventsData, GetAssetEventsResponse, CreateAssetEventData,
CreateAssetEventResponse, MaterializeAssetData, MaterializeAssetResponse,
GetAssetQueuedEventsData, GetAssetQueuedEventsResponse,
DeleteAssetQueuedEventsData, DeleteAssetQueuedEventsResponse, GetAssetData,
GetAssetResponse, GetDagAssetQueuedEventsData, GetDagAssetQueuedEventsResponse,
Dele [...]
+import type { GetAssetsData, GetAssetsResponse, GetAssetAliasesData,
GetAssetAliasesResponse, GetAssetAliasData, GetAssetAliasResponse,
GetAssetEventsData, GetAssetEventsResponse, CreateAssetEventData,
CreateAssetEventResponse, MaterializeAssetData, MaterializeAssetResponse,
GetAssetQueuedEventsData, GetAssetQueuedEventsResponse,
DeleteAssetQueuedEventsData, DeleteAssetQueuedEventsResponse, GetAssetData,
GetAssetResponse, GetDagAssetQueuedEventsData, GetDagAssetQueuedEventsResponse,
Dele [...]
export class AssetService {
/**
@@ -407,6 +407,57 @@ export class AssetService {
});
}
+ /**
+ * Get Assets
+ * Get assets. Like the public endpoint, but also supports sorting by
group and last asset event timestamp.
+ * @param data The data for the request.
+ * @param data.limit
+ * @param data.offset
+ * @param data.namePattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `name_prefix_pattern` on large tables — see "Filtering with pattern
parameters".
+ * @param data.namePrefixPattern Case-sensitive, index-friendly prefix
match. See "Filtering with pattern parameters".
+ * @param data.uri Exact-match filter on the full asset URI. Compiles to
an indexed equality comparison (``uri = ...``). Repeat the parameter
(``?uri=a&uri=b``) to match multiple assets.
+ * @param data.uriPattern Case-insensitive substring match (SQL `ILIKE`).
Slower than `uri_prefix_pattern` on large tables — see "Filtering with pattern
parameters".
+ * @param data.uriPrefixPattern Case-sensitive, index-friendly prefix
match. See "Filtering with pattern parameters".
+ * @param data.groupPattern Case-insensitive substring match (SQL
`ILIKE`). Slower than `group_prefix_pattern` on large tables — see "Filtering
with pattern parameters".
+ * @param data.groupPrefixPattern Case-sensitive, index-friendly prefix
match. See "Filtering with pattern parameters".
+ * @param data.dagIds
+ * @param data.onlyActive
+ * @param data.lastAssetEventTimestampGte
+ * @param data.lastAssetEventTimestampGt
+ * @param data.lastAssetEventTimestampLte
+ * @param data.lastAssetEventTimestampLt
+ * @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
name, uri, group, created_at, updated_at, last_asset_event_timestamp`
+ * @returns AssetCollectionResponse Successful Response
+ * @throws ApiError
+ */
+ public static getAssetsUi(data: GetAssetsUiData = {}):
CancelablePromise<GetAssetsUiResponse> {
+ return __request(OpenAPI, {
+ method: 'GET',
+ url: '/ui/assets',
+ query: {
+ limit: data.limit,
+ offset: data.offset,
+ name_pattern: data.namePattern,
+ name_prefix_pattern: data.namePrefixPattern,
+ uri: data.uri,
+ uri_pattern: data.uriPattern,
+ uri_prefix_pattern: data.uriPrefixPattern,
+ group_pattern: data.groupPattern,
+ group_prefix_pattern: data.groupPrefixPattern,
+ dag_ids: data.dagIds,
+ only_active: data.onlyActive,
+ last_asset_event_timestamp_gte:
data.lastAssetEventTimestampGte,
+ last_asset_event_timestamp_gt: data.lastAssetEventTimestampGt,
+ last_asset_event_timestamp_lte:
data.lastAssetEventTimestampLte,
+ last_asset_event_timestamp_lt: data.lastAssetEventTimestampLt,
+ order_by: data.orderBy
+ },
+ errors: {
+ 422: 'Validation Error'
+ }
+ });
+ }
+
/**
* Next Run Assets
* @param data The data for the request.
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 932326c9d6f..2c6a870e0e3 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
@@ -2988,6 +2988,51 @@ export type DeleteDagAssetQueuedEventData = {
export type DeleteDagAssetQueuedEventResponse = void;
+export type GetAssetsUiData = {
+ dagIds?: Array<(string)>;
+ /**
+ * Case-insensitive substring match (SQL `ILIKE`). Slower than
`group_prefix_pattern` on large tables — see "Filtering with pattern
parameters".
+ */
+ groupPattern?: string | null;
+ /**
+ * Case-sensitive, index-friendly prefix match. See "Filtering with
pattern parameters".
+ */
+ groupPrefixPattern?: string | null;
+ lastAssetEventTimestampGt?: string | null;
+ lastAssetEventTimestampGte?: string | null;
+ lastAssetEventTimestampLt?: string | null;
+ lastAssetEventTimestampLte?: string | null;
+ limit?: number;
+ /**
+ * Case-insensitive substring match (SQL `ILIKE`). Slower than
`name_prefix_pattern` on large tables — see "Filtering with pattern parameters".
+ */
+ namePattern?: string | null;
+ /**
+ * Case-sensitive, index-friendly prefix match. See "Filtering with
pattern parameters".
+ */
+ namePrefixPattern?: string | null;
+ offset?: number;
+ onlyActive?: boolean;
+ /**
+ * Attributes to order by, multi criteria sort is supported. Prefix with
`-` for descending order. Supported attributes: `id, name, uri, group,
created_at, updated_at, last_asset_event_timestamp`
+ */
+ orderBy?: Array<(string)>;
+ /**
+ * Exact-match filter on the full asset URI. Compiles to an indexed
equality comparison (``uri = ...``). Repeat the parameter (``?uri=a&uri=b``) to
match multiple assets.
+ */
+ uri?: Array<(string)>;
+ /**
+ * Case-insensitive substring match (SQL `ILIKE`). Slower than
`uri_prefix_pattern` on large tables — see "Filtering with pattern parameters".
+ */
+ uriPattern?: string | null;
+ /**
+ * Case-sensitive, index-friendly prefix match. See "Filtering with
pattern parameters".
+ */
+ uriPrefixPattern?: string | null;
+};
+
+export type GetAssetsUiResponse = AssetCollectionResponse;
+
export type NextRunAssetsData = {
dagId: string;
};
@@ -5117,6 +5162,21 @@ export type $OpenApiTs = {
};
};
};
+ '/ui/assets': {
+ get: {
+ req: GetAssetsUiData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 200: AssetCollectionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ };
'/ui/next_run_assets/{dag_id}': {
get: {
req: NextRunAssetsData;
diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/assets.json
b/airflow-core/src/airflow/ui/public/i18n/locales/en/assets.json
index a5d044e6264..0b963c05f2a 100644
--- a/airflow-core/src/airflow/ui/public/i18n/locales/en/assets.json
+++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/assets.json
@@ -42,7 +42,10 @@
"title": "Create Asset Event for {{name}}"
},
"events": "Events",
- "extra": "Extra",
+ "filters": {
+ "groupPlaceholder": "Search group",
+ "lastEventDateRange": "Last Event Date"
+ },
"group": "Group",
"lastAssetEvent": "Last Asset Event",
"name": "Name",
diff --git a/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx
b/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx
index 7c29e276d7b..d29df264d34 100644
--- a/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx
+++ b/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx
@@ -61,7 +61,7 @@ export enum FilterTypes {
}
export const useFilterConfigs = () => {
- const { t: translate } = useTranslation(["browse", "common", "components",
"admin", "hitl"]);
+ const { t: translate } = useTranslation(["assets", "browse", "common",
"components", "admin", "hitl"]);
const multiTeamEnabled = Boolean(useConfig("multi_team"));
const { data: teamsData } = useTeamsServiceListTeams({ orderBy: ["name"] },
undefined, {
enabled: multiTeamEnabled,
@@ -179,6 +179,14 @@ export const useFilterConfigs = () => {
label: translate("admin:jobs.columns.executorClass"),
type: FilterTypes.TEXT,
},
+ [SearchParamsKeys.GROUP_PATTERN]: {
+ hotkeyDisabled: true,
+ icon: <FiDatabase />,
+ label: translate("assets:group"),
+ placeholder: translate("assets:filters.groupPlaceholder"),
+ supportsAdvancedSearch: true,
+ type: FilterTypes.TEXT,
+ },
[SearchParamsKeys.HOSTNAME]: {
hotkeyDisabled: true,
icon: <MdComputer />,
@@ -209,6 +217,13 @@ export const useFilterConfigs = () => {
supportsAdvancedSearch: true,
type: FilterTypes.TEXT,
},
+ [SearchParamsKeys.LAST_ASSET_EVENT_TIMESTAMP_RANGE]: {
+ endKey: SearchParamsKeys.LAST_ASSET_EVENT_TIMESTAMP_LTE,
+ icon: <MdDateRange />,
+ label: translate("assets:filters.lastEventDateRange"),
+ startKey: SearchParamsKeys.LAST_ASSET_EVENT_TIMESTAMP_GTE,
+ type: FilterTypes.DATERANGE,
+ },
[SearchParamsKeys.LOGICAL_DATE_RANGE]: {
endKey: SearchParamsKeys.LOGICAL_DATE_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 71f2b763e8a..b2ac5532911 100644
--- a/airflow-core/src/airflow/ui/src/constants/searchParams.ts
+++ b/airflow-core/src/airflow/ui/src/constants/searchParams.ts
@@ -54,11 +54,15 @@ export enum SearchParamsKeys {
GRAPH_OPERATOR = "graph-operator",
GRAPH_TASK_GROUP = "graph-task_group",
GRAPH_TASK_STATE = "graph-task_state",
+ GROUP_PATTERN = "group_pattern",
HOSTNAME = "hostname",
INCLUDED_EVENTS = "included_events",
JOB_STATE = "job_state",
JOB_TYPE = "job_type",
KEY_PATTERN = "key_pattern",
+ LAST_ASSET_EVENT_TIMESTAMP_GTE = "last_asset_event_timestamp_gte",
+ LAST_ASSET_EVENT_TIMESTAMP_LTE = "last_asset_event_timestamp_lte",
+ LAST_ASSET_EVENT_TIMESTAMP_RANGE = "last_asset_event_timestamp_range",
LAST_DAG_RUN_STATE = "last_dag_run_state",
LIMIT = "limit",
LOG_LEVEL = "log_level",
diff --git a/airflow-core/src/airflow/ui/src/pages/AssetsList/AssetsList.tsx
b/airflow-core/src/airflow/ui/src/pages/AssetsList/AssetsList.tsx
index 150b57ba968..33518105231 100644
--- a/airflow-core/src/airflow/ui/src/pages/AssetsList/AssetsList.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/AssetsList/AssetsList.tsx
@@ -16,34 +16,35 @@
* specific language governing permissions and limitations
* under the License.
*/
-import { Flex, Heading, useDisclosure, VStack } from "@chakra-ui/react";
+import { Heading, VStack } from "@chakra-ui/react";
import type { ColumnDef } from "@tanstack/react-table";
import { useTranslation } from "react-i18next";
import { useSearchParams } from "react-router-dom";
-import { useAssetServiceGetAssets } from "openapi/queries";
+import { useAssetServiceGetAssetsUi } from "openapi/queries";
import type { AssetResponse } from "openapi/requests/types.gen";
import { DataTable } from "src/components/DataTable";
import { useTableURLState } from "src/components/DataTable/useTableUrlState";
import { ErrorAlert } from "src/components/ErrorAlert";
-import { ExpandCollapseButtons } from "src/components/ExpandCollapseButtons";
-import RenderedJsonField from "src/components/RenderedJsonField";
+import { FilterBar } from "src/components/FilterBar";
import { SearchBar } from "src/components/SearchBar";
import Time from "src/components/Time";
import { RouterLink } from "src/components/ui";
import { SearchParamsKeys, type SearchParamsKeysType } from
"src/constants/searchParams";
-import { useAdvancedSearch } from "src/hooks/useAdvancedSearch";
+import { useAdvancedSearch, useAdvancedSearchArg } from
"src/hooks/useAdvancedSearch";
import { CreateAssetEvent } from "src/pages/Asset/CreateAssetEvent";
-import { useDocumentTitle } from "src/utils";
+import { useDocumentTitle, useFiltersHandler, type FilterableSearchParamsKeys
} from "src/utils";
import { DependencyPopover } from "./DependencyPopover";
+const assetsFilterKeys: Array<FilterableSearchParamsKeys> = [
+ SearchParamsKeys.GROUP_PATTERN,
+ SearchParamsKeys.LAST_ASSET_EVENT_TIMESTAMP_RANGE,
+];
+
type AssetRow = { row: { original: AssetResponse } };
-const createColumns = (
- translate: (key: string) => string,
- open?: boolean,
-): Array<ColumnDef<AssetResponse>> => [
+const createColumns = (translate: (key: string) => string):
Array<ColumnDef<AssetResponse>> => [
{
accessorKey: "name",
cell: ({ row: { original } }: AssetRow) => (
@@ -54,7 +55,7 @@ const createColumns = (
header: () => translate("name"),
},
{
- accessorKey: "last_asset_event",
+ accessorKey: "last_asset_event_timestamp",
cell: ({ row: { original } }: AssetRow) => {
const assetEvent = original.last_asset_event;
const timestamp = assetEvent?.timestamp;
@@ -65,12 +66,10 @@ const createColumns = (
return <Time datetime={timestamp} />;
},
- enableSorting: false,
header: () => translate("lastAssetEvent"),
},
{
accessorKey: "group",
- enableSorting: false,
header: () => translate("group"),
},
{
@@ -97,21 +96,6 @@ const createColumns = (
enableSorting: false,
header: "",
},
- {
- accessorKey: "extra",
- cell: ({ row: { original } }) => {
- if (original.extra !== null) {
- return <RenderedJsonField collapsed={!open} content={original.extra ??
{}} />;
- }
-
- return undefined;
- },
- enableSorting: false,
- header: translate("extra"),
- meta: {
- skeletonWidth: 200,
- },
- },
];
const { NAME_PATTERN, OFFSET }: SearchParamsKeysType = SearchParamsKeys;
@@ -129,18 +113,31 @@ export const AssetsList = () => {
const { setTableURLState, tableURLState } = useTableURLState();
const { pagination, sorting } = tableURLState;
const [sort] = sorting;
- const orderBy = sort ? [`${sort.desc ? "-" : ""}${sort.id}`] : undefined;
+ const orderBy = sort ? [`${sort.desc ? "-" : ""}${sort.id}`] :
["-last_asset_event_timestamp"];
- const { onClose, onOpen, open } = useDisclosure();
+ const { filterConfigs, handleFiltersChange, initialValues } =
useFiltersHandler(assetsFilterKeys);
- const { data, error, isLoading } = useAssetServiceGetAssets({
+ const lastAssetEventTimestampGte =
searchParams.get(SearchParamsKeys.LAST_ASSET_EVENT_TIMESTAMP_GTE);
+ const lastAssetEventTimestampLte =
searchParams.get(SearchParamsKeys.LAST_ASSET_EVENT_TIMESTAMP_LTE);
+ const groupArg = useAdvancedSearchArg({
+ patternApiKey: "groupPattern",
+ prefixApiKey: "groupPrefixPattern",
+ storageKey: SearchParamsKeys.GROUP_PATTERN,
+ value: searchParams.get(SearchParamsKeys.GROUP_PATTERN),
+ });
+
+ const { data, error, isLoading } = useAssetServiceGetAssetsUi({
+ ...groupArg,
+ lastAssetEventTimestampGte: lastAssetEventTimestampGte ?? undefined,
+ lastAssetEventTimestampLte: lastAssetEventTimestampLte ?? undefined,
limit: pagination.pageSize,
...(advancedSearch.enabled ? { namePattern } : { namePrefixPattern:
namePattern }),
offset: pagination.pageIndex * pagination.pageSize,
orderBy,
});
- const columns = createColumns(translate, open);
+ const columns = createColumns(translate);
+ const totalEntries = data?.total_entries ?? 0;
const handleSearchChange = (value: string) => {
setTableURLState({
@@ -166,18 +163,15 @@ export const AssetsList = () => {
placeholder={translate("searchPlaceholder")}
/>
- <Flex alignItems="center" justifyContent="space-between">
- <Heading py={3} size="md">
- {data?.total_entries} {translate("common:asset", { count:
data?.total_entries })}
- </Heading>
- <ExpandCollapseButtons
- collapseLabel={translate("common:collapseAllExtra")}
- expandLabel={translate("common:expandAllExtra")}
- isExpanded={open}
- onCollapse={onClose}
- onExpand={onOpen}
- />
- </Flex>
+ <FilterBar
+ configs={filterConfigs}
+ initialValues={initialValues}
+ onFiltersChange={handleFiltersChange}
+ />
+
+ <Heading py={3} size="md">
+ {totalEntries} {translate("common:asset", { count: totalEntries })}
+ </Heading>
</VStack>
<DataTable
columns={columns}
@@ -188,7 +182,7 @@ export const AssetsList = () => {
modelName="common:asset"
onStateChange={setTableURLState}
showRowCountHeading={false}
- total={data?.total_entries}
+ total={totalEntries}
/>
</>
);
diff --git a/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts
b/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts
index 54f71369370..9349249c078 100644
--- a/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts
+++ b/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts
@@ -73,10 +73,12 @@ export type FilterableSearchParamsKeys =
| SearchParamsKeys.EVENT_DATE_RANGE
| SearchParamsKeys.EVENT_TYPE
| SearchParamsKeys.EXECUTOR_CLASS
+ | SearchParamsKeys.GROUP_PATTERN
| SearchParamsKeys.HOSTNAME
| SearchParamsKeys.JOB_STATE
| SearchParamsKeys.JOB_TYPE
| SearchParamsKeys.KEY_PATTERN
+ | SearchParamsKeys.LAST_ASSET_EVENT_TIMESTAMP_RANGE
| SearchParamsKeys.LOGICAL_DATE_RANGE
| SearchParamsKeys.MAP_INDEX
| SearchParamsKeys.MISSED
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 810b809a131..05735154c5f 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
@@ -23,12 +23,15 @@ import pendulum
import pytest
from sqlalchemy import select
+from airflow.models import DagModel
from airflow.models.asset import (
AssetActive,
+ AssetAliasModel,
AssetDagRunQueue,
AssetEvent,
AssetModel,
AssetPartitionDagRun,
+ DagScheduleAssetReference,
PartitionedAssetKeyLog,
)
from airflow.partition_mappers.base import RollupMapper
@@ -39,7 +42,13 @@ 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.db import clear_db_apdr, clear_db_dags,
clear_db_pakl, clear_db_serialized_dags
+from tests_common.test_utils.db import (
+ clear_db_apdr,
+ clear_db_assets,
+ clear_db_dags,
+ clear_db_pakl,
+ clear_db_serialized_dags,
+)
pytestmark = pytest.mark.db_test
@@ -471,3 +480,278 @@ class TestNextRunAssets:
body = response.json()
assert len(body["events"]) == 1
assert body["events"][0]["asset_inactive"] is True
+
+
+class TestGetAssetsUi:
+ @pytest.fixture(autouse=True)
+ def cleanup_assets(self):
+ clear_db_assets()
+
+ yield
+
+ clear_db_assets()
+
+ def test_should_respond_401(self, unauthenticated_test_client):
+ response = unauthenticated_test_client.get("/assets")
+ assert response.status_code == 401
+
+ def test_should_respond_403(self, unauthorized_test_client):
+ response = unauthorized_test_client.get("/assets")
+ assert response.status_code == 403
+
+ def test_should_respond_200(self, test_client, session):
+ asset = AssetModel(name="ui_asset", uri="s3://bucket/ui_asset",
group="asset")
+ session.add(asset)
+ session.add(AssetActive.for_asset(asset))
+ session.commit()
+
+ response = test_client.get("/assets")
+ assert response.status_code == 200
+ body = response.json()
+ assert body["total_entries"] == 1
+ assert body["assets"][0]["name"] == "ui_asset"
+
+ def test_sort_by_last_asset_event_timestamp(self, test_client, session):
+ older = AssetModel(name="older", uri="s3://bucket/older",
group="asset")
+ newer = AssetModel(name="newer", uri="s3://bucket/newer",
group="asset")
+ session.add_all([older, newer])
+ session.add(AssetActive.for_asset(older))
+ session.add(AssetActive.for_asset(newer))
+ session.flush()
+
+ base = pendulum.datetime(2024, 1, 1)
+ session.add(AssetEvent(asset_id=older.id, timestamp=base))
+ session.add(AssetEvent(asset_id=newer.id, timestamp=base.add(days=1)))
+ session.commit()
+
+ response =
test_client.get("/assets?order_by=last_asset_event_timestamp")
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] == ["older",
"newer"]
+
+ response =
test_client.get("/assets?order_by=-last_asset_event_timestamp")
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] == ["newer",
"older"]
+
+ def test_default_sort_is_last_asset_event_timestamp_desc(self,
test_client, session):
+ older = AssetModel(name="older", uri="s3://bucket/older_default",
group="asset")
+ newer = AssetModel(name="newer", uri="s3://bucket/newer_default",
group="asset")
+ session.add_all([older, newer])
+ session.add(AssetActive.for_asset(older))
+ session.add(AssetActive.for_asset(newer))
+ session.flush()
+
+ base = pendulum.datetime(2024, 1, 1)
+ session.add(AssetEvent(asset_id=older.id, timestamp=base))
+ session.add(AssetEvent(asset_id=newer.id, timestamp=base.add(days=1)))
+ session.commit()
+
+ response = test_client.get("/assets")
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] == ["newer",
"older"]
+
+ def test_sort_by_group(self, test_client, session):
+ billing = AssetModel(name="billing_asset",
uri="s3://bucket/billing_sort", group="billing")
+ marketing = AssetModel(name="marketing_asset",
uri="s3://bucket/marketing_sort", group="marketing")
+ session.add_all([billing, marketing])
+ session.add(AssetActive.for_asset(billing))
+ session.add(AssetActive.for_asset(marketing))
+ session.commit()
+
+ response = test_client.get("/assets?order_by=group")
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] ==
["billing_asset", "marketing_asset"]
+
+ response = test_client.get("/assets?order_by=-group")
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] ==
["marketing_asset", "billing_asset"]
+
+ def test_filter_by_group_pattern(self, test_client, session):
+ billing = AssetModel(name="billing_asset", uri="s3://bucket/billing",
group="billing")
+ marketing = AssetModel(name="marketing_asset",
uri="s3://bucket/marketing", group="marketing")
+ session.add_all([billing, marketing])
+ session.add(AssetActive.for_asset(billing))
+ session.add(AssetActive.for_asset(marketing))
+ session.commit()
+
+ response = test_client.get("/assets?group_pattern=bill")
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] ==
["billing_asset"]
+
+ def test_filter_by_group_prefix_pattern(self, test_client, session):
+ billing = AssetModel(name="billing_asset",
uri="s3://bucket/billing_prefix", group="billing")
+ rebilling = AssetModel(name="rebilling_asset",
uri="s3://bucket/rebilling", group="rebilling")
+ session.add_all([billing, rebilling])
+ session.add(AssetActive.for_asset(billing))
+ session.add(AssetActive.for_asset(rebilling))
+ session.commit()
+
+ # Prefix match anchors at the start, so "bill" excludes "rebilling"
(substring would not).
+ response = test_client.get("/assets?group_prefix_pattern=bill")
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] ==
["billing_asset"]
+
+ def test_filter_by_last_asset_event_timestamp_range(self, test_client,
session):
+ older = AssetModel(name="older", uri="s3://bucket/older_range",
group="asset")
+ newer = AssetModel(name="newer", uri="s3://bucket/newer_range",
group="asset")
+ session.add_all([older, newer])
+ session.add(AssetActive.for_asset(older))
+ session.add(AssetActive.for_asset(newer))
+ session.flush()
+
+ base = pendulum.datetime(2024, 1, 1)
+ session.add(AssetEvent(asset_id=older.id, timestamp=base))
+ session.add(AssetEvent(asset_id=newer.id, timestamp=base.add(days=10)))
+ session.commit()
+
+ response = test_client.get(
+ "/assets", params={"last_asset_event_timestamp_gte":
base.add(days=5).isoformat()}
+ )
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] == ["newer"]
+
+ def test_aliases_present_for_asset_via_alias(self, test_client, session):
+ """
+ Regression test for https://github.com/apache/airflow/issues/58058:
+ aliases must be visible via the Assets endpoints, not just fetchable
+ through the CLI/DB.
+ """
+ asset = AssetModel(name="alias_target",
uri="s3://bucket/alias_target", group="asset")
+ alias = AssetAliasModel(name="my-alias", group="")
+ session.add_all([asset, alias])
+ session.flush()
+ session.add(AssetActive.for_asset(asset))
+ asset.aliases.append(alias)
+ session.commit()
+
+ response = test_client.get("/assets")
+ assert response.status_code == 200
+ body = response.json()
+ assert len(body["assets"]) == 1
+ assert body["assets"][0]["aliases"] == [{"id": alias.id, "name":
"my-alias", "group": ""}]
+
+ def test_total_entries_counts_assets_not_events(self, test_client,
session):
+ """The outer join to AssetEvent must not inflate total_entries for
assets with many events."""
+ multi = AssetModel(name="multi", uri="s3://bucket/multi",
group="reporting")
+ solo = AssetModel(name="solo", uri="s3://bucket/solo",
group="reporting")
+ other = AssetModel(name="other", uri="s3://bucket/other", group="ops")
+ session.add_all([multi, solo, other])
+ for asset in (multi, solo, other):
+ session.add(AssetActive.for_asset(asset))
+ session.flush()
+
+ base = pendulum.datetime(2024, 1, 1)
+ for offset in range(3):
+ session.add(AssetEvent(asset_id=multi.id,
timestamp=base.add(hours=offset)))
+ session.add(AssetEvent(asset_id=solo.id, timestamp=base))
+ session.commit()
+
+ response = test_client.get("/assets")
+ assert response.status_code == 200
+ body = response.json()
+ assert body["total_entries"] == 3
+ assert sorted(a["name"] for a in body["assets"]) == ["multi", "other",
"solo"]
+
+ response = test_client.get("/assets?group_pattern=report")
+ assert response.status_code == 200
+ body = response.json()
+ assert body["total_entries"] == 2
+ assert sorted(a["name"] for a in body["assets"]) == ["multi", "solo"]
+
+ def test_pagination_covers_all_assets(self, test_client, session):
+ """Paging returns every asset exactly once with a stable total, mixing
evented and never-evented."""
+ evented_new = AssetModel(name="evented_new", uri="s3://bucket/en",
group="asset")
+ evented_old = AssetModel(name="evented_old", uri="s3://bucket/eo",
group="asset")
+ never_a = AssetModel(name="never_a", uri="s3://bucket/na",
group="asset")
+ never_b = AssetModel(name="never_b", uri="s3://bucket/nb",
group="asset")
+ session.add_all([evented_new, evented_old, never_a, never_b])
+ for asset in (evented_new, evented_old, never_a, never_b):
+ session.add(AssetActive.for_asset(asset))
+ session.flush()
+
+ base = pendulum.datetime(2024, 1, 1)
+ session.add(AssetEvent(asset_id=evented_old.id, timestamp=base))
+ session.add(AssetEvent(asset_id=evented_new.id,
timestamp=base.add(days=1)))
+ session.commit()
+
+ page_one = test_client.get("/assets?limit=2&offset=0")
+ page_two = test_client.get("/assets?limit=2&offset=2")
+ assert page_one.status_code == 200
+ assert page_two.status_code == 200
+ assert page_one.json()["total_entries"] == 4
+ assert page_two.json()["total_entries"] == 4
+
+ names = [a["name"] for a in page_one.json()["assets"] +
page_two.json()["assets"]]
+ assert sorted(names) == ["evented_new", "evented_old", "never_a",
"never_b"]
+
+ def test_timestamp_range_filter_excludes_never_evented_assets(self,
test_client, session):
+ """A timestamp range filter excludes assets with no event (NULL fails
the range predicate)."""
+ evented = AssetModel(name="evented", uri="s3://bucket/re",
group="asset")
+ never = AssetModel(name="never", uri="s3://bucket/rn", group="asset")
+ session.add_all([evented, never])
+ session.add(AssetActive.for_asset(evented))
+ session.add(AssetActive.for_asset(never))
+ session.flush()
+
+ base = pendulum.datetime(2024, 1, 1)
+ session.add(AssetEvent(asset_id=evented.id, timestamp=base))
+ session.commit()
+
+ response = test_client.get(
+ "/assets", params={"last_asset_event_timestamp_gte":
base.subtract(days=1).isoformat()}
+ )
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] == ["evented"]
+
+ def test_filter_by_only_active(self, test_client, session):
+ active = AssetModel(name="active", uri="s3://bucket/active",
group="asset")
+ inactive = AssetModel(name="inactive", uri="s3://bucket/inactive",
group="asset")
+ session.add_all([active, inactive])
+ session.add(AssetActive.for_asset(active))
+ session.commit()
+
+ response = test_client.get("/assets")
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] == ["active"]
+
+ response = test_client.get("/assets?only_active=false")
+ assert response.status_code == 200
+ assert sorted(a["name"] for a in response.json()["assets"]) ==
["active", "inactive"]
+
+ def test_filter_by_uri_pattern(self, test_client, session):
+ s3 = AssetModel(name="s3_asset", uri="s3://bucket/key", group="asset")
+ gcs = AssetModel(name="gcs_asset", uri="gcs://bucket/key",
group="asset")
+ session.add_all([s3, gcs])
+ session.add(AssetActive.for_asset(s3))
+ session.add(AssetActive.for_asset(gcs))
+ session.commit()
+
+ response = test_client.get("/assets?uri_pattern=s3")
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] == ["s3_asset"]
+
+ @pytest.mark.usefixtures("testing_dag_bundle")
+ def test_filter_by_dag_ids(self, test_client, session):
+ referenced = AssetModel(name="referenced",
uri="s3://bucket/referenced", group="asset")
+ unreferenced = AssetModel(name="unreferenced",
uri="s3://bucket/unreferenced", group="asset")
+ session.add_all([referenced, unreferenced])
+ session.add(AssetActive.for_asset(referenced))
+ session.add(AssetActive.for_asset(unreferenced))
+ session.add(DagModel(dag_id="consumer_dag", bundle_name="testing"))
+ session.add(DagScheduleAssetReference(dag_id="consumer_dag",
asset=referenced))
+ session.commit()
+
+ response = test_client.get("/assets?dag_ids=consumer_dag")
+ assert response.status_code == 200
+ assert [a["name"] for a in response.json()["assets"]] == ["referenced"]
+
+ def test_query_count(self, test_client, session):
+ """The asset relationships are eager-loaded, so the query count stays
fixed regardless of
+ how many assets are returned (a lazy-loading regression would issue
queries per asset)."""
+ for i in range(5):
+ asset = AssetModel(name=f"asset{i}", uri=f"s3://bucket/asset{i}",
group="asset")
+ session.add(asset)
+ session.add(AssetActive.for_asset(asset))
+ session.commit()
+
+ with assert_queries_count(7):
+ assert test_client.get("/assets").status_code == 200
