You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "bbovenzi (via GitHub)" <gi...@apache.org> on 2026/07/15 21:29:41 UTC
[PR] Updated sort+filters for Assets List [airflow]
bbovenzi opened a new pull request, #69940:
URL: https://github.com/apache/airflow/pull/69940
Create a custom ui/assets list endpoint to give us more control. Such as:
- sort by last asset event
- default the sort by last asset event
- filter by group search
- filter by last asset event time range
- sorting by group
Also removed asset extra from the list page. We only need to show it on the asset details page
<img width="1112" height="623" alt="Screenshot 2026-07-15 at 5 29 02 PM" src="https://github.com/user-attachments/assets/cd97564d-70f4-46dd-a89f-c2e3c5d5fcab" />
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [x] Yes Clause Sonnet 5
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Updated sort+filters for Assets List [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #69940:
URL: https://github.com/apache/airflow/pull/69940#discussion_r3646673720
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -499,14 +499,10 @@ def search_param_factory(
else "The pipe `|` is matched literally, not as an OR separator. "
)
DESCRIPTION = (
- "SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). "
+ "Case-insensitive substring match; `%` / `_` wildcards allowed. "
f"{pipe_clause}"
- "Regular expressions are **not** supported. "
- "\n\n"
- "**Performance note:** this full-match pattern is evaluated as ``ILIKE '%term%'`` and "
- "most of the time prevents the database from using B-tree indexes, which can be very "
- "slow on large tables. Prefer the equivalent "
- f"``{pattern_name.replace('_pattern', '_prefix_pattern')}`` parameter when possible."
+ f"Not index-friendly — prefer ``{pattern_name.replace('_pattern', '_prefix_pattern')}`` "
+ "on large tables."
Review Comment:
For both descriptoin update here I prefered the original one, which is more explicit.
That's not related directly to the PR maybe we should keep the original, it will also reduce the diff cause all generated search have this description embedded.
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py:
##########
@@ -471,3 +480,315 @@ def test_non_partitioned_asset_inactive_true_when_deactivated(self, test_client,
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):
Review Comment:
Can we add some `assert_queries_count` guards just to make sure we don't have N+1 queries problem (by forgetting an eager loading or explicit join etc..)
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py:
##########
@@ -49,6 +71,83 @@
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,
+ )
+
+ # CASE key keeps assets with no event (NULL timestamp) last in both directions.
+ order_columns: list[ColumnElement] = []
Review Comment:
I would keep things consistent with other APIs, nulls behavior is decided by the backend implementation (db).
Also a case prevent the leverage of indexes, so performance wise it's not great, that's why we removed a very similar piece of code from. the `parameters.py` module.
##########
airflow-core/src/airflow/ui/src/pages/AssetsList/AssetsList.tsx:
##########
@@ -126,21 +110,36 @@ export const AssetsList = () => {
const namePattern = searchParams.get(NAME_PATTERN) ?? "";
const advancedSearch = useAdvancedSearch("assets");
- const { setTableURLState, tableURLState } = useTableURLState();
+ const { setTableURLState, tableURLState } = useTableURLState({
+ sorting: [{ desc: true, id: "last_asset_event_timestamp" }],
+ });
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),
+ });
+
Review Comment:
We are missing the ability to switch from substring to prefix search for search fields:
<img width="349" height="183" alt="Image" src="https://github.com/user-attachments/assets/fe1ff53a-7af5-4d58-905d-269541c3b773" />
<img width="389" height="173" alt="Image" src="https://github.com/user-attachments/assets/e6ec2b88-02d8-4fb2-8ca4-7af6efccbbeb" />
##########
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"))
Review Comment:
This changes the `max(timestamp)` to `max(id)` that's behavior change for the public endpoint. Is that expected? Should we keep the former?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Updated sort+filters for Assets List [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun merged PR #69940:
URL: https://github.com/apache/airflow/pull/69940
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Updated sort+filters for Assets List [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #69940:
URL: https://github.com/apache/airflow/pull/69940#discussion_r3681678465
##########
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),
Review Comment:
Each of these five `subqueryload`s re-runs the whole parent statement as an inner subquery, `GROUP BY asset_event` and all. I dumped the executed SQL for a 2-row page and got six statements, five of which contain `SELECT asset_event.asset_id, max(asset_event.id) ... GROUP BY asset_event.asset_id`; with the count query that is seven aggregations of `asset_event` for one list page. `selectinload` emits a plain `WHERE asset_id IN (<page ids>)` instead and is already what the sibling helper in `common/db/dags.py` uses, so is there a reason not to switch here?
##########
airflow-core/src/airflow/ui/src/pages/AssetsList/AssetsList.tsx:
##########
@@ -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({
Review Comment:
This changes the key the list caches under, and `CreateAssetEventModal`'s success handler only invalidates `GetAssetEvents` plus the dag run and task instance keys. It never invalidated the old assets key either, so the play button on a row already left the timestamp stale, but now that last event is the default sort the row also stays in the wrong position. Adding `[useAssetServiceGetAssetsUiKey]` to that `queryKeys` array would cover both, alongside the `[useDagServiceGetDagsUiKey]` that is already there.
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py:
##########
@@ -471,3 +480,278 @@ def test_non_partitioned_asset_inactive_true_when_deactivated(self, test_client,
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):
Review Comment:
None of these five assets has a watcher, alias, scheduled dag or task, so every eager-loaded collection comes back empty and the nested `joinedload(AssetWatcherModel.trigger)` never fires. I checked by deleting that `.joinedload(...)` from `generate_assets_with_last_event_query`, and all 17 tests in this class still pass at 7, even though `from_asset_row` would then lazy-load `watcher.trigger` once per watcher. Could the fixture give at least one asset a watcher and an alias, so the guard covers the relationships it exists to protect?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Updated sort+filters for Assets List [airflow]
Posted by "bbovenzi (via GitHub)" <gi...@apache.org>.
bbovenzi commented on PR #69940:
URL: https://github.com/apache/airflow/pull/69940#issuecomment-4996720763
> > cc: @gopidesupavan I removed extra from the Assets list. After your work on data quality is that ok, or should we keep it?
>
> Yeah it should be fine.. i guess if we click on the asset and that will show us additional data field?
Yes, its still on the details page.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Updated sort+filters for Assets List [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on PR #69940:
URL: https://github.com/apache/airflow/pull/69940#issuecomment-4996127296
> cc: @gopidesupavan I removed extra from the Assets list. After your work on data quality is that ok, or should we keep it?
Yeah it should be fine.. i guess if we click on the asset and that will show us additional data field?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Updated sort+filters for Assets List [airflow]
Posted by "bbovenzi (via GitHub)" <gi...@apache.org>.
bbovenzi commented on PR #69940:
URL: https://github.com/apache/airflow/pull/69940#issuecomment-4996057144
cc: @gopidesupavan I removed extra from the Assets list. After your work on data quality is that ok, or should we keep it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org