You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kaxil (via GitHub)" <gi...@apache.org> on 2026/09/10 23:10:40 UTC
[PR] Show Dag bundle versions and refresh times in the UI [airflow]
kaxil opened a new pull request, #72909:
URL: https://github.com/apache/airflow/pull/72909
## Problem
A Dag author pushes a commit and has no way to tell whether Airflow has picked it up. The workflow today is: commit, push, then reload the Dags page until the import errors or the file details change. Where Dag code viewing is disabled there is not even that, and the author is left guessing the bundle refresh interval.
Nothing in the UI or the REST API exposes Dag bundle state, though `dag_bundle` has recorded the version and last refresh since Airflow 3.0.
## Solution
`GET /api/v2/dagBundles` lists each bundle with the version Airflow currently holds, when a Dag processor last refreshed it, how many Dag import errors it has, its owning team, and whether it is still configured. A **Dag Bundles** page under *Browse* renders that and refreshes itself while open.
For a Git bundle `version` is the commit SHA of the tracking ref, comparable directly against what was deployed, which also makes "wait until my commit is live" scriptable from a deployment pipeline. `last_refreshed` advances on every successful refresh even when the version did not change, so it doubles as "a processor is alive and looking at this bundle". A non-zero `import_error_count` beside a version that just moved means the code was fetched but a Dag in it failed to import.
Read-only on purpose: no way to create, delete, enable or disable a bundle, so no new role and no new writable configuration surface. That keeps it clear of the objection on #61645, which is about runtime *mutation* of bundle configuration.
## Authorization, the decision I would most like a second opinion on
A bundle is visible to a caller who can read at least one Dag recorded against it. The route carries `requires_access_dag(method="GET")` and rows are filtered by `PermittedDagBundleFilter`, the existing `PermittedDagFilter` pattern applied to `dag_bundle` through the Dags inside it. `import_error_count` is counted on the same terms as `GET /importErrors`, and `bundle_url` is withheld from callers who may not read Dag versions.
I deliberately did not add an `AccessView.DAG_BUNDLES`. It is the house style for admin-ish read views, but it needs a FAB `RESOURCE_DAG_BUNDLE` seeded into roles, and seeding that admin- or op-only locks out the Dag authors this exists for.
I also rejected `is_authorized_dag(method="GET", details=DagDetails(id=None, team_name=...))`. FAB reads that as "may this user list Dags", which `_is_authorized_list_dags` answers True for anyone holding read on a single Dag, and FAB ignores `team_name` outright, so one Dag's worth of access would have exposed every bundle's repository URL. Deriving from readable Dags behaves identically across FAB, SimpleAuthManager and Keycloak.
#72715 faces the same question and asked it without an answer, so I am raising it rather than treating either choice as settled.
## Known limitations
- It cannot tell you a refresh is *failing*: a raising `bundle.refresh()` is caught and logged, leaving `last_refreshed` at its last success with no error marker, so "failing for an hour" looks like "not due yet". Documented, with the two checks that need no refresh interval. A `last_refresh_error` column would fix it properly and is follow-up work.
- A new version does not mean the new Dags are running; the bundle is refreshed before its Dags are parsed.
- A bundle from which no Dag has ever parsed successfully is not listed at all.
Screenshots and a short demo of the page updating live are being added to this description.
Related: #61645, #72715, #60616.
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #72909:
URL: https://github.com/apache/airflow/pull/72909#discussion_r4020261496
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_bundles.py:
##########
@@ -0,0 +1,216 @@
+# 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, Annotated
+
+from fastapi import Depends
+from sqlalchemy import and_, func, select
+
+from airflow.api_fastapi.auth.managers.models.resource_details import AccessView, DagAccessEntity
+from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.dag_bundles import (
+ DagBundleCollectionResponse,
+ DagBundleResponse,
+)
+from airflow.api_fastapi.core_api.security import (
+ AuthManagerDep,
+ GetUserDep,
+ ReadableDagBundlesFilterDep,
+ requires_access_dag,
+)
+from airflow.configuration import conf
+from airflow.models import DagModel
+from airflow.models.dagbundle import DagBundleModel
+from airflow.models.errors import ParseImportError
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from sqlalchemy.orm import Session
+
+ from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager
+ from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
+
+dag_bundles_router = AirflowRouter(tags=["Dag Bundle"], prefix="/dagBundles")
+
+
+def _import_error_counts(
+ *,
+ bundle_names: Sequence[str],
+ readable_dag_ids: set[str],
+ auth_manager: BaseAuthManager,
+ user: BaseUser,
+ session: Session,
+) -> dict[str, int] | None:
+ """
+ Count the import errors per bundle that this caller is allowed to know about.
+
+ Reproduces the two-part authorization of ``GET /importErrors`` rather than counting every row
+ for the bundle: an error in a file the caller can read no Dag in stays hidden, and an error in
+ a file with no registered Dag needs the admin-by-default ``IMPORT_ERRORS_ALL``, since the
+ file's existence would otherwise leak. Returns ``None`` when the caller may not read import
+ errors at all.
+ """
+ if not auth_manager.authorize_view(access_view=AccessView.IMPORT_ERRORS, user=user):
+ return None
+ if not bundle_names:
+ return {}
+
+ # Files -- keyed ``(relative_fileloc, bundle_name)`` -- in which the caller can read a Dag.
+ readable_files = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(
+ DagModel.dag_id.in_(readable_dag_ids),
+ DagModel.bundle_name.in_(bundle_names),
+ )
+ .distinct()
+ .subquery()
+ )
+ counts: dict[str, int] = {
+ bundle_name: count
+ for bundle_name, count in session.execute(
+ select(ParseImportError.bundle_name, func.count())
+ .join(
+ readable_files,
+ and_(
+ ParseImportError.filename == readable_files.c.relative_fileloc,
+ ParseImportError.bundle_name == readable_files.c.bundle_name,
+ ),
+ )
+ .group_by(ParseImportError.bundle_name)
+ ).all()
+ if bundle_name is not None
+ }
+
+ # Errors for files that never registered a Dag, added only where the caller holds the
+ # admin-by-default view for that bundle's team. Narrowed to this page's bundles, unlike the
+ # equivalent in ``import_error.py``: this endpoint polls, so an unbounded ``SELECT DISTINCT``
+ # over ``dag`` every few seconds would be a real cost on a large deployment.
+ files_with_any_dags = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(DagModel.bundle_name.in_(bundle_names))
+ .distinct()
+ .subquery()
+ )
+ unregistered = session.execute(
+ select(ParseImportError.bundle_name, func.count())
+ .outerjoin(
+ files_with_any_dags,
+ and_(
+ ParseImportError.filename == files_with_any_dags.c.relative_fileloc,
+ ParseImportError.bundle_name == files_with_any_dags.c.bundle_name,
+ ),
+ )
+ .where(
+ files_with_any_dags.c.relative_fileloc.is_(None),
+ ParseImportError.bundle_name.in_(bundle_names),
+ )
+ .group_by(ParseImportError.bundle_name)
+ ).all()
+ if unregistered:
+ team_names = DagBundleModel.get_team_names(
+ [bundle_name for bundle_name, _ in unregistered if bundle_name], session=session
+ )
+ for bundle_name, count in unregistered:
+ if bundle_name is None:
+ continue
+ if auth_manager.authorize_view(
+ access_view=AccessView.IMPORT_ERRORS_ALL,
+ user=user,
+ team_name=team_names.get(bundle_name),
+ ):
+ counts[bundle_name] = counts.get(bundle_name, 0) + count
+
+ return counts
+
+
+@dag_bundles_router.get("", dependencies=[Depends(requires_access_dag(method="GET"))])
+def get_dag_bundles(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["name", "version", "last_refreshed", "active"], DagBundleModel).dynamic_depends()),
+ ],
+ readable_dag_bundles_filter: ReadableDagBundlesFilterDep,
+ auth_manager: AuthManagerDep,
+ session: SessionDep,
+ user: GetUserDep,
+) -> DagBundleCollectionResponse:
+ """
+ List the Dag bundles Airflow knows about.
+
+ Reports the version of each bundle Airflow currently holds and when a Dag processor last
+ refreshed it, so whoever deployed the code can tell whether it has been picked up yet.
+
+ A bundle is visible to a user who can read at least one Dag recorded against it. A bundle
+ holding no registered Dag at all has nothing to authorize against, so it is listed only for
+ a user holding the admin-by-default ``AccessView.IMPORT_ERRORS_ALL`` -- the same view that
+ governs import errors for a file that never registered a Dag.
+ """
+ bundles_select, total_entries = paginated_select(
+ statement=select(DagBundleModel),
+ filters=[readable_dag_bundles_filter],
+ order_by=order_by,
+ offset=offset,
+ limit=limit,
+ session=session,
+ )
+ bundles = session.scalars(bundles_select).all()
+ bundle_names = [bundle.name for bundle in bundles]
+
+ # ``DagBundleModel.team_name`` walks a lazily loaded relationship, so read them in one query.
+ team_names = (
+ DagBundleModel.get_team_names(bundle_names, session=session)
+ if bundle_names and conf.getboolean("core", "multi_team")
+ else {}
+ )
+
+ import_error_counts = _import_error_counts(
+ bundle_names=bundle_names,
+ readable_dag_ids=readable_dag_bundles_filter.value or set(),
+ auth_manager=auth_manager,
+ user=user,
+ session=session,
+ )
+
+ # A rendered bundle url is otherwise only reachable through ``GET /dags/{dag_id}/dagVersions``,
+ # which additionally requires Dag *version* read, so withhold it from a role without that.
+ show_bundle_url = auth_manager.is_authorized_dag(
Review Comment:
Added in 8e53572 as `test_withholds_the_url_without_dag_version_access`. It patches `is_authorized_dag` to deny only `DagAccessEntity.VERSION`, and asserts the bundle is still listed, with `bundle_url` null and the rest of the row intact, since dropping the row would defeat the point of the page.
Confirmed it bites: replacing the guard with `show_bundle_url = True` fails this test and nothing else, so it was the missing case exactly as you said.
On the two duplicate import-error tests, you are right that they were one test written twice. Both used the same client, made the same request and asserted the same number, so neither isolated the half its docstring claimed. They are now one test that says what each of the three seeded errors is for and why the count is 1, and notes that dropping either restriction takes it to 2, so a single assertion pins both halves.
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #72909:
URL: https://github.com/apache/airflow/pull/72909#discussion_r4020244469
##########
airflow-core/src/airflow/api_fastapi/core_api/security.py:
##########
@@ -398,6 +438,34 @@ def depends_permitted_dags_filter(
]
+def readable_dag_bundles_filter_factory() -> Callable[[BaseUser, BaseAuthManager], PermittedDagBundleFilter]:
+ """
+ Create a callable for Depends in FastAPI that returns the Dag bundle filter for the user.
+
+ Dag bundles need their own factory rather than ``permitted_dag_filter_factory``: besides the
+ readable Dag ids, the filter needs to know whether the user may see bundles that hold no
+ registered Dag, which is a separate authorization decision.
+ """
+
+ def depends_readable_dag_bundles_filter(
+ user: GetUserDep,
+ auth_manager: AuthManagerDep,
+ ) -> PermittedDagBundleFilter:
+ return PermittedDagBundleFilter(
+ auth_manager.get_authorized_dag_ids(user=user, method="GET"),
+ include_dagless_bundles=auth_manager.authorize_view(
+ access_view=AccessView.IMPORT_ERRORS_ALL, user=user
Review Comment:
You are right, and thank you for reproducing both directions. Fixed in 8e53572.
The Dag-less bundle names are now resolved before pagination, their teams read with the batched `DagBundleModel.get_team_names`, and each one authorized with `authorize_view(IMPORT_ERRORS_ALL, team_name=...)`. The survivors go into the query as an explicit allow-list, so counting and pagination see the same set, which is the shape `get_import_errors` already uses for a file with no registered Dag.
Both of your cases are now regression tests in `TestDaglessBundleTeamScoping`. I checked they actually bite: reverting the filter to the single unscoped check fails three of the four, with exactly the two failures you named, over-exposing `team_b` and hiding `team_a`.
The extra cost is one `SELECT name FROM dag_bundle WHERE name NOT IN (SELECT bundle_name FROM dag)` per request plus the team lookup, which is the trade you called for and matches `/importErrors`. It does land on the polling endpoint, which is the same surface as your index comment, so the follow-up there covers this query too.
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil merged PR #72909:
URL: https://github.com/apache/airflow/pull/72909
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #72909:
URL: https://github.com/apache/airflow/pull/72909#issuecomment-5689852630
### Backport failed to create: airflow-ctl/v0-1-test. View the failure log <a href='https://github.com/apache/airflow/actions/runs/35038134363'> Run details </a>
Note: As of [Merging PRs targeted for Airflow 3.X](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#merging-prs-targeted-for-airflow-3x)
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.
In matter of doubt please ask in [#release-management](https://apache-airflow.slack.com/archives/C03G9H97MM2) Slack channel.
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>❌</td>
<td>airflow-ctl/v0-1-test</td>
<td><a href="https://github.com/apache/airflow/commit/d57b249abeea54a6a2bb850faadcdde65b140cc5"><img src='https://img.shields.io/badge/Commit-d57b249-red' alt='Commit Link'></a></td>
</tr>
</table>
You can attempt to backport this manually by running:
```bash
cherry_picker d57b249 airflow-ctl/v0-1-test
```
This should apply the commit to the airflow-ctl/v0-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.
After you have resolved the conflicts, you can continue the backport process by running:
```bash
cherry_picker --continue
```
If you don't have cherry-picker installed, see the [installation guide](https://github.com/apache/airflow/blob/main/dev/README.md#how-to-backport-pr-with-cherry-picker-cli).
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #72909:
URL: https://github.com/apache/airflow/pull/72909#discussion_r4007948174
##########
airflow-core/src/airflow/api_fastapi/core_api/security.py:
##########
@@ -361,6 +336,21 @@ def to_orm(self, select: Select) -> Select:
return select.where(Backfill.dag_id.in_(self.value or set()))
+class PermittedDagBundleFilter(PermittedDagFilter):
+ """A parameter that filters Dag bundles to the ones holding a Dag the user may read."""
+
+ def to_orm(self, statement: Select) -> Select:
+ # A bundle carries no per-Dag key to authorize on, so it is scoped by the Dags inside it.
+ # Filtering in the query keeps unauthorized rows out of the count and pagination as well.
+ # A bundle from which no Dag has ever parsed is therefore invisible until one does, and one
+ # whose Dags have since been removed stays visible while a stale ``DagModel`` row names it.
+ return statement.where(
+ DagBundleModel.name.in_(
+ select(DagModel.bundle_name).where(DagModel.dag_id.in_(self.value or set()))
+ )
+ )
Review Comment:
Agreed, and fixed in 5c3248b.
`PermittedDagBundleFilter` now takes `include_dagless_bundles`, set from `authorize_view(AccessView.IMPORT_ERRORS_ALL)` in a new `readable_dag_bundles_filter_factory` modelled on `readable_event_logs_filter_factory`. A bundle with no `DagModel` row has nothing to authorize against, so it rides on the same admin-by-default view that already governs import errors for a file that never registered a Dag. The added clause is `OR dag_bundle.name NOT IN (SELECT dag.bundle_name ...)`, additive over the readable-Dag disjunct, so it never bypasses Dag scoping for a bundle that does have Dags, and it stays in SQL so pagination and `total_entries` are still correct.
One piece I would like your read on. The view is authorized once without a `team_name`, because which bundles qualify is not known until the query runs. That is asymmetric with `_import_error_counts`, which does scope per bundle team. It is inert under FAB and SimpleAuthManager since both treat admin as global, but a team-aware manager granting the unscoped view would show a holder every Dag-less bundle rather than only their own teams'. There is a comment at the call site saying so.
If you would rather it were fully team-scoped, the shape is: resolve the Dag-less bundle names first, filter them with `authorize_view(..., team_name=...)`, and pass the survivors into the filter as an allow-list. That costs one extra query per request on an endpoint that polls, which is the cost you flag in your other comment, so I did not want to choose that unilaterally.
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #72909:
URL: https://github.com/apache/airflow/pull/72909#discussion_r4014489330
##########
airflow-core/src/airflow/api_fastapi/core_api/security.py:
##########
@@ -398,6 +438,34 @@ def depends_permitted_dags_filter(
]
+def readable_dag_bundles_filter_factory() -> Callable[[BaseUser, BaseAuthManager], PermittedDagBundleFilter]:
+ """
+ Create a callable for Depends in FastAPI that returns the Dag bundle filter for the user.
+
+ Dag bundles need their own factory rather than ``permitted_dag_filter_factory``: besides the
+ readable Dag ids, the filter needs to know whether the user may see bundles that hold no
+ registered Dag, which is a separate authorization decision.
+ """
+
+ def depends_readable_dag_bundles_filter(
+ user: GetUserDep,
+ auth_manager: AuthManagerDep,
+ ) -> PermittedDagBundleFilter:
+ return PermittedDagBundleFilter(
+ auth_manager.get_authorized_dag_ids(user=user, method="GET"),
+ include_dagless_bundles=auth_manager.authorize_view(
+ access_view=AccessView.IMPORT_ERRORS_ALL, user=user
Review Comment:
This needs the per-team allow-list. With a team-aware policy, allowing the unscoped view while denying team B still exposes B's bundle; denying the unscoped view while allowing team A hides A's bundle. Both cases were reproduced, and `/importErrors` handles them correctly with the same policy. Please authorize against each bundle's actual team before pagination and counting. We can batch and reuse the team lookup.
##########
airflow-core/src/airflow/api_fastapi/common/types.py:
##########
@@ -120,6 +120,7 @@ class MenuItem(Enum):
CONFIG = "Config"
CONNECTIONS = "Connections"
DAGS = "Dags"
+ DAG_BUNDLES = "Dag Bundles"
Review Comment:
Keycloak's `TEAM_MENU_ITEMS` and `TEAM_ADMIN_MENU_ITEMS` still omit this new item. Team `Viewer/User/Op/Admin` roles therefore cannot discover the page through Browse despite being able to read their bundles through the API. Add the menu permission with the provider's compatibility guard and cover the generated grant.
##########
airflow-core/src/airflow/ui/tests/e2e/specs/dag-bundles.spec.ts:
##########
@@ -0,0 +1,71 @@
+/*!
+ * 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.
+ */
+import { expect, test } from "tests/e2e/fixtures";
+
+test.describe("Dag Bundles Page", () => {
+ test.beforeEach(async ({ dagBundlesPage }) => {
+ await dagBundlesPage.navigate();
+ });
+
+ test("verify dag bundles page heading", async ({ dagBundlesPage }) => {
+ await expect(dagBundlesPage.heading).toBeVisible();
+ });
+
+ test("Verify Dag Bundles page is accessible via Browse menu", async ({ dagBundlesPage }) => {
+ await dagBundlesPage.navigateFromBrowseMenu();
+
+ await dagBundlesPage.waitForLoad();
+ await expect(dagBundlesPage.heading).toBeVisible();
+ expect(await dagBundlesPage.getRowCount()).toBeGreaterThan(0);
+ });
+
+ test("Verify the dag bundles list displays", async ({ dagBundlesPage }) => {
+ await expect(dagBundlesPage.table).toBeVisible();
+ });
+
+ test("Verify the configured bundle is listed with a name and an active state", async ({
+ dagBundlesPage,
+ }) => {
+ expect(await dagBundlesPage.getRowCount()).toBeGreaterThan(0);
+
+ await expect(dagBundlesPage.nameCellAt(0)).not.toBeEmpty();
+ // Spelled out both ways rather than left blank when healthy, so the cell is never empty.
+ await expect(dagBundlesPage.activeCellAt(0)).toHaveText(/^(active|inactive)$/i);
+ });
+
+ test("Verify the version cell says what it knows", async ({ dagBundlesPage }) => {
Review Comment:
These tests only check the initial render, so removing automatic refresh would leave them green. Please add fake-timer Vitest coverage for the ten-second minimum, longer configured intervals, zero disabling refresh, and an updated response changing the displayed row. The Browse navigation check can remain an e2e test.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_bundles.py:
##########
@@ -0,0 +1,216 @@
+# 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, Annotated
+
+from fastapi import Depends
+from sqlalchemy import and_, func, select
+
+from airflow.api_fastapi.auth.managers.models.resource_details import AccessView, DagAccessEntity
+from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.dag_bundles import (
+ DagBundleCollectionResponse,
+ DagBundleResponse,
+)
+from airflow.api_fastapi.core_api.security import (
+ AuthManagerDep,
+ GetUserDep,
+ ReadableDagBundlesFilterDep,
+ requires_access_dag,
+)
+from airflow.configuration import conf
+from airflow.models import DagModel
+from airflow.models.dagbundle import DagBundleModel
+from airflow.models.errors import ParseImportError
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from sqlalchemy.orm import Session
+
+ from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager
+ from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
+
+dag_bundles_router = AirflowRouter(tags=["Dag Bundle"], prefix="/dagBundles")
+
+
+def _import_error_counts(
+ *,
+ bundle_names: Sequence[str],
+ readable_dag_ids: set[str],
+ auth_manager: BaseAuthManager,
+ user: BaseUser,
+ session: Session,
+) -> dict[str, int] | None:
+ """
+ Count the import errors per bundle that this caller is allowed to know about.
+
+ Reproduces the two-part authorization of ``GET /importErrors`` rather than counting every row
+ for the bundle: an error in a file the caller can read no Dag in stays hidden, and an error in
+ a file with no registered Dag needs the admin-by-default ``IMPORT_ERRORS_ALL``, since the
+ file's existence would otherwise leak. Returns ``None`` when the caller may not read import
+ errors at all.
+ """
+ if not auth_manager.authorize_view(access_view=AccessView.IMPORT_ERRORS, user=user):
+ return None
+ if not bundle_names:
+ return {}
+
+ # Files -- keyed ``(relative_fileloc, bundle_name)`` -- in which the caller can read a Dag.
+ readable_files = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(
+ DagModel.dag_id.in_(readable_dag_ids),
+ DagModel.bundle_name.in_(bundle_names),
+ )
+ .distinct()
+ .subquery()
+ )
+ counts: dict[str, int] = {
+ bundle_name: count
+ for bundle_name, count in session.execute(
+ select(ParseImportError.bundle_name, func.count())
+ .join(
+ readable_files,
+ and_(
+ ParseImportError.filename == readable_files.c.relative_fileloc,
+ ParseImportError.bundle_name == readable_files.c.bundle_name,
+ ),
+ )
+ .group_by(ParseImportError.bundle_name)
+ ).all()
+ if bundle_name is not None
+ }
+
+ # Errors for files that never registered a Dag, added only where the caller holds the
+ # admin-by-default view for that bundle's team. Narrowed to this page's bundles, unlike the
+ # equivalent in ``import_error.py``: this endpoint polls, so an unbounded ``SELECT DISTINCT``
+ # over ``dag`` every few seconds would be a real cost on a large deployment.
+ files_with_any_dags = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(DagModel.bundle_name.in_(bundle_names))
+ .distinct()
+ .subquery()
+ )
+ unregistered = session.execute(
+ select(ParseImportError.bundle_name, func.count())
+ .outerjoin(
+ files_with_any_dags,
+ and_(
+ ParseImportError.filename == files_with_any_dags.c.relative_fileloc,
+ ParseImportError.bundle_name == files_with_any_dags.c.bundle_name,
+ ),
+ )
+ .where(
+ files_with_any_dags.c.relative_fileloc.is_(None),
+ ParseImportError.bundle_name.in_(bundle_names),
+ )
+ .group_by(ParseImportError.bundle_name)
+ ).all()
+ if unregistered:
+ team_names = DagBundleModel.get_team_names(
+ [bundle_name for bundle_name, _ in unregistered if bundle_name], session=session
+ )
+ for bundle_name, count in unregistered:
+ if bundle_name is None:
+ continue
+ if auth_manager.authorize_view(
+ access_view=AccessView.IMPORT_ERRORS_ALL,
+ user=user,
+ team_name=team_names.get(bundle_name),
+ ):
+ counts[bundle_name] = counts.get(bundle_name, 0) + count
+
+ return counts
+
+
+@dag_bundles_router.get("", dependencies=[Depends(requires_access_dag(method="GET"))])
+def get_dag_bundles(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["name", "version", "last_refreshed", "active"], DagBundleModel).dynamic_depends()),
+ ],
+ readable_dag_bundles_filter: ReadableDagBundlesFilterDep,
+ auth_manager: AuthManagerDep,
+ session: SessionDep,
+ user: GetUserDep,
+) -> DagBundleCollectionResponse:
+ """
+ List the Dag bundles Airflow knows about.
+
+ Reports the version of each bundle Airflow currently holds and when a Dag processor last
+ refreshed it, so whoever deployed the code can tell whether it has been picked up yet.
+
+ A bundle is visible to a user who can read at least one Dag recorded against it. A bundle
+ holding no registered Dag at all has nothing to authorize against, so it is listed only for
+ a user holding the admin-by-default ``AccessView.IMPORT_ERRORS_ALL`` -- the same view that
+ governs import errors for a file that never registered a Dag.
+ """
+ bundles_select, total_entries = paginated_select(
+ statement=select(DagBundleModel),
+ filters=[readable_dag_bundles_filter],
+ order_by=order_by,
+ offset=offset,
+ limit=limit,
+ session=session,
+ )
+ bundles = session.scalars(bundles_select).all()
+ bundle_names = [bundle.name for bundle in bundles]
+
+ # ``DagBundleModel.team_name`` walks a lazily loaded relationship, so read them in one query.
+ team_names = (
+ DagBundleModel.get_team_names(bundle_names, session=session)
+ if bundle_names and conf.getboolean("core", "multi_team")
+ else {}
+ )
+
+ import_error_counts = _import_error_counts(
+ bundle_names=bundle_names,
+ readable_dag_ids=readable_dag_bundles_filter.value or set(),
+ auth_manager=auth_manager,
+ user=user,
+ session=session,
+ )
+
+ # A rendered bundle url is otherwise only reachable through ``GET /dags/{dag_id}/dagVersions``,
+ # which additionally requires Dag *version* read, so withhold it from a role without that.
+ show_bundle_url = auth_manager.is_authorized_dag(
Review Comment:
Can we add a test where the caller can read the Dag but is denied `DagAccessEntity.VERSION`, and verify that the bundle remains visible with `bundle_url: null`. The current tests all allow version access, so removing this guard would not fail them.
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #72909:
URL: https://github.com/apache/airflow/pull/72909#discussion_r4020255155
##########
airflow-core/src/airflow/ui/tests/e2e/specs/dag-bundles.spec.ts:
##########
@@ -0,0 +1,71 @@
+/*!
+ * 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.
+ */
+import { expect, test } from "tests/e2e/fixtures";
+
+test.describe("Dag Bundles Page", () => {
+ test.beforeEach(async ({ dagBundlesPage }) => {
+ await dagBundlesPage.navigate();
+ });
+
+ test("verify dag bundles page heading", async ({ dagBundlesPage }) => {
+ await expect(dagBundlesPage.heading).toBeVisible();
+ });
+
+ test("Verify Dag Bundles page is accessible via Browse menu", async ({ dagBundlesPage }) => {
+ await dagBundlesPage.navigateFromBrowseMenu();
+
+ await dagBundlesPage.waitForLoad();
+ await expect(dagBundlesPage.heading).toBeVisible();
+ expect(await dagBundlesPage.getRowCount()).toBeGreaterThan(0);
+ });
+
+ test("Verify the dag bundles list displays", async ({ dagBundlesPage }) => {
+ await expect(dagBundlesPage.table).toBeVisible();
+ });
+
+ test("Verify the configured bundle is listed with a name and an active state", async ({
+ dagBundlesPage,
+ }) => {
+ expect(await dagBundlesPage.getRowCount()).toBeGreaterThan(0);
+
+ await expect(dagBundlesPage.nameCellAt(0)).not.toBeEmpty();
+ // Spelled out both ways rather than left blank when healthy, so the cell is never empty.
+ await expect(dagBundlesPage.activeCellAt(0)).toHaveText(/^(active|inactive)$/i);
+ });
+
+ test("Verify the version cell says what it knows", async ({ dagBundlesPage }) => {
Review Comment:
Agreed, they were smoke checks. Added `src/pages/DagBundles/DagBundles.test.tsx` in 8e53572 with fake timers over the real page and `msw`, covering the four cases you asked for: the ten-second floor holding against the 3s default, a configured 30s being left alone rather than pulled down, 0 disabling refresh entirely, and an updated response changing the rendered version.
I verified they fail for the right reason rather than trusting them. Replacing `refetchInterval` with `configuredInterval` (dropping the floor and the zero guard) fails the floor test; replacing it with `false` fails three of the four, which is the regression you described.
Two things worth knowing about the harness. The page fetches once before `[api] auto_refresh_interval` lands and again once it does, so each test drains that startup traffic and then counts from the settled total instead of from zero. And StrictMode mounts the page twice, so a tick issues one request per observer; the assertions are therefore "refreshed" or "did not refresh" rather than an exact delta.
The Browse navigation check stays in Playwright, as you suggested.
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #72909:
URL: https://github.com/apache/airflow/pull/72909#discussion_r4004022606
##########
airflow-core/docs/administration-and-deployment/dag-bundles.rst:
##########
@@ -226,6 +226,39 @@ Starting Airflow 3.0.2 git is pre installed in the base image. However, if you a
ENV GIT_PYTHON_REFRESH=quiet
+Monitoring Dag bundles
+----------------------
+
+The **Dag Bundles** page under *Browse* shows, for each bundle, the version Airflow currently holds,
+when a Dag processor last refreshed it, its import-error count, and whether it is still configured
+-- so you can tell whether a commit has been picked up without reading the Dag source. The page
+refreshes itself while open, and the same data is available at ``GET /api/v2/dagBundles`` for a
+deployment pipeline to poll.
+
+``version`` is whatever the bundle reports; for a Git Dag bundle it is the commit SHA of the tracking
+ref. Bundles that do not support versioning, such as ``LocalDagBundle``, report ``null``.
+
+``last_refreshed`` is the last *successful* refresh and advances whether or not the version changed.
+The cadence is the bundle's :ref:`config:dag_processor__refresh_interval`, checked every
+:ref:`config:dag_processor__bundle_refresh_check_interval`.
+
+Some caveats:
+
+* A failing refresh looks like one that is not due, since a refresh that raises is logged and leaves
+ ``last_refreshed`` at its last success. Check the Dag processor logs, and
+ ``/api/v2/jobs?job_type=DagProcessorJob`` for a live processor.
+* A new version does not mean the new Dags are running, because bundles are refreshed before their
+ Dags are parsed -- read the import-error count alongside it.
Review Comment:
Reading the import-error count alongside the version still cannot tell us whether that commit parsed successfully. The version advances before parsing, error rows aren't versioned, and parser timeouts only update in-memory stats, so even an admin can keep seeing a fresh SHA with zero errors while parsing fails. Ordinary readers can also see zero when errors are filtered out by permissions. We can make the unknown/partial state explicit in the UI and avoid presenting this combination as a deployment-health check.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_bundles.py:
##########
@@ -0,0 +1,214 @@
+# 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, Annotated
+
+from fastapi import Depends
+from sqlalchemy import and_, func, select
+
+from airflow.api_fastapi.auth.managers.models.resource_details import AccessView, DagAccessEntity
+from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.dag_bundles import (
+ DagBundleCollectionResponse,
+ DagBundleResponse,
+)
+from airflow.api_fastapi.core_api.security import (
+ AuthManagerDep,
+ GetUserDep,
+ ReadableDagBundlesFilterDep,
+ requires_access_dag,
+)
+from airflow.configuration import conf
+from airflow.models import DagModel
+from airflow.models.dagbundle import DagBundleModel
+from airflow.models.errors import ParseImportError
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from sqlalchemy.orm import Session
+
+ from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager
+ from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
+
+dag_bundles_router = AirflowRouter(tags=["Dag Bundle"], prefix="/dagBundles")
+
+
+def _import_error_counts(
+ *,
+ bundle_names: Sequence[str],
+ readable_dag_ids: set[str],
+ auth_manager: BaseAuthManager,
+ user: BaseUser,
+ session: Session,
+) -> dict[str, int] | None:
+ """
+ Count the import errors per bundle that this caller is allowed to know about.
+
+ Reproduces the two-part authorization of ``GET /importErrors`` rather than counting every row
+ for the bundle: an error in a file the caller can read no Dag in stays hidden, and an error in
+ a file with no registered Dag needs the admin-by-default ``IMPORT_ERRORS_ALL``, since the
+ file's existence would otherwise leak. Returns ``None`` when the caller may not read import
+ errors at all.
+ """
+ if not auth_manager.authorize_view(access_view=AccessView.IMPORT_ERRORS, user=user):
+ return None
+ if not bundle_names:
+ return {}
+
+ # Files -- keyed ``(relative_fileloc, bundle_name)`` -- in which the caller can read a Dag.
+ readable_files = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(
+ DagModel.dag_id.in_(readable_dag_ids),
+ DagModel.bundle_name.in_(bundle_names),
+ )
+ .distinct()
+ .subquery()
+ )
+ counts: dict[str, int] = {
+ bundle_name: count
+ for bundle_name, count in session.execute(
+ select(ParseImportError.bundle_name, func.count())
+ .join(
+ readable_files,
+ and_(
+ ParseImportError.filename == readable_files.c.relative_fileloc,
+ ParseImportError.bundle_name == readable_files.c.bundle_name,
+ ),
+ )
+ .group_by(ParseImportError.bundle_name)
+ ).all()
+ if bundle_name is not None
+ }
+
+ # Errors for files that never registered a Dag, added only where the caller holds the
+ # admin-by-default view for that bundle's team. Narrowed to this page's bundles, unlike the
+ # equivalent in ``import_error.py``: this endpoint polls, so an unbounded ``SELECT DISTINCT``
+ # over ``dag`` every few seconds would be a real cost on a large deployment.
+ files_with_any_dags = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(DagModel.bundle_name.in_(bundle_names))
Review Comment:
Filtering to this page's bundle names doesn't avoid the table scans here: `dag.bundle_name` and the import-error bundle/file lookup have no supporting indexes, and the query plan still scans both tables. Every open page repeats this work every ten seconds, including computing unregistered errors that authorization later discards. Maybe as a follow up, can we add the required indexes and skip that work for bundles whose unregistered errors the caller cannot read?
##########
airflow-core/src/airflow/api_fastapi/core_api/security.py:
##########
@@ -361,6 +336,21 @@ def to_orm(self, select: Select) -> Select:
return select.where(Backfill.dag_id.in_(self.value or set()))
+class PermittedDagBundleFilter(PermittedDagFilter):
+ """A parameter that filters Dag bundles to the ones holding a Dag the user may read."""
+
+ def to_orm(self, statement: Select) -> Select:
+ # A bundle carries no per-Dag key to authorize on, so it is scoped by the Dags inside it.
+ # Filtering in the query keeps unauthorized rows out of the count and pagination as well.
+ # A bundle from which no Dag has ever parsed is therefore invisible until one does, and one
+ # whose Dags have since been removed stays visible while a stale ``DagModel`` row names it.
+ return statement.where(
+ DagBundleModel.name.in_(
+ select(DagModel.bundle_name).where(DagModel.dag_id.in_(self.value or set()))
+ )
+ )
Review Comment:
This hides a new bundle even from an admin when its first parse fails, because there is no `DagModel` row yet. I don't think successful parsing should be a prerequisite for monitoring a bundle. We need a bundle/team or privileged authorization path that can show these rows without a registered Dag.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_bundles.py:
##########
@@ -0,0 +1,214 @@
+# 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, Annotated
+
+from fastapi import Depends
+from sqlalchemy import and_, func, select
+
+from airflow.api_fastapi.auth.managers.models.resource_details import AccessView, DagAccessEntity
+from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.dag_bundles import (
+ DagBundleCollectionResponse,
+ DagBundleResponse,
+)
+from airflow.api_fastapi.core_api.security import (
+ AuthManagerDep,
+ GetUserDep,
+ ReadableDagBundlesFilterDep,
+ requires_access_dag,
+)
+from airflow.configuration import conf
+from airflow.models import DagModel
+from airflow.models.dagbundle import DagBundleModel
+from airflow.models.errors import ParseImportError
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from sqlalchemy.orm import Session
+
+ from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager
+ from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
+
+dag_bundles_router = AirflowRouter(tags=["Dag Bundle"], prefix="/dagBundles")
+
+
+def _import_error_counts(
+ *,
+ bundle_names: Sequence[str],
+ readable_dag_ids: set[str],
+ auth_manager: BaseAuthManager,
+ user: BaseUser,
+ session: Session,
+) -> dict[str, int] | None:
+ """
+ Count the import errors per bundle that this caller is allowed to know about.
+
+ Reproduces the two-part authorization of ``GET /importErrors`` rather than counting every row
+ for the bundle: an error in a file the caller can read no Dag in stays hidden, and an error in
+ a file with no registered Dag needs the admin-by-default ``IMPORT_ERRORS_ALL``, since the
+ file's existence would otherwise leak. Returns ``None`` when the caller may not read import
+ errors at all.
+ """
+ if not auth_manager.authorize_view(access_view=AccessView.IMPORT_ERRORS, user=user):
+ return None
+ if not bundle_names:
+ return {}
+
+ # Files -- keyed ``(relative_fileloc, bundle_name)`` -- in which the caller can read a Dag.
+ readable_files = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(
+ DagModel.dag_id.in_(readable_dag_ids),
+ DagModel.bundle_name.in_(bundle_names),
+ )
+ .distinct()
+ .subquery()
+ )
+ counts: dict[str, int] = {
+ bundle_name: count
+ for bundle_name, count in session.execute(
+ select(ParseImportError.bundle_name, func.count())
+ .join(
+ readable_files,
+ and_(
+ ParseImportError.filename == readable_files.c.relative_fileloc,
+ ParseImportError.bundle_name == readable_files.c.bundle_name,
+ ),
+ )
+ .group_by(ParseImportError.bundle_name)
+ ).all()
+ if bundle_name is not None
+ }
+
+ # Errors for files that never registered a Dag, added only where the caller holds the
+ # admin-by-default view for that bundle's team. Narrowed to this page's bundles, unlike the
+ # equivalent in ``import_error.py``: this endpoint polls, so an unbounded ``SELECT DISTINCT``
+ # over ``dag`` every few seconds would be a real cost on a large deployment.
+ files_with_any_dags = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(DagModel.bundle_name.in_(bundle_names))
+ .distinct()
+ .subquery()
+ )
+ unregistered = session.execute(
+ select(ParseImportError.bundle_name, func.count())
+ .outerjoin(
+ files_with_any_dags,
+ and_(
+ ParseImportError.filename == files_with_any_dags.c.relative_fileloc,
+ ParseImportError.bundle_name == files_with_any_dags.c.bundle_name,
+ ),
+ )
+ .where(
+ files_with_any_dags.c.relative_fileloc.is_(None),
+ ParseImportError.bundle_name.in_(bundle_names),
+ )
+ .group_by(ParseImportError.bundle_name)
+ ).all()
+ if unregistered:
+ team_names = DagBundleModel.get_team_names(
+ [bundle_name for bundle_name, _ in unregistered if bundle_name], session=session
+ )
+ for bundle_name, count in unregistered:
+ if bundle_name is None:
+ continue
+ if auth_manager.authorize_view(
+ access_view=AccessView.IMPORT_ERRORS_ALL,
+ user=user,
+ team_name=team_names.get(bundle_name),
+ ):
+ counts[bundle_name] = counts.get(bundle_name, 0) + count
+
+ return counts
+
+
+@dag_bundles_router.get("", dependencies=[Depends(requires_access_dag(method="GET"))])
+def get_dag_bundles(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["name", "version", "last_refreshed", "active"], DagBundleModel).dynamic_depends()),
+ ],
+ readable_dag_bundles_filter: ReadableDagBundlesFilterDep,
+ auth_manager: AuthManagerDep,
+ session: SessionDep,
+ user: GetUserDep,
+) -> DagBundleCollectionResponse:
+ """
+ List the Dag bundles Airflow knows about.
+
+ Reports the version of each bundle Airflow currently holds and when a Dag processor last
+ refreshed it, so whoever deployed the code can tell whether it has been picked up yet.
+
+ A bundle is visible to a user who can read at least one Dag recorded against it, so one from
+ which no Dag has ever parsed successfully is not listed at all.
+ """
+ bundles_select, total_entries = paginated_select(
+ statement=select(DagBundleModel),
+ filters=[readable_dag_bundles_filter],
+ order_by=order_by,
+ offset=offset,
+ limit=limit,
+ session=session,
+ )
+ bundles = session.scalars(bundles_select).all()
+ bundle_names = [bundle.name for bundle in bundles]
+
+ # ``DagBundleModel.team_name`` walks a lazily loaded relationship, so read them in one query.
+ team_names = (
+ DagBundleModel.get_team_names(bundle_names, session=session)
+ if bundle_names and conf.getboolean("core", "multi_team")
+ else {}
+ )
+
+ import_error_counts = _import_error_counts(
+ bundle_names=bundle_names,
+ readable_dag_ids=readable_dag_bundles_filter.value or set(),
+ auth_manager=auth_manager,
+ user=user,
+ session=session,
+ )
+
+ # A rendered bundle url is otherwise only reachable through ``GET /dags/{dag_id}/dagVersions``,
+ # which additionally requires Dag *version* read, so withhold it from a role without that.
+ show_bundle_url = auth_manager.is_authorized_dag(
+ method="GET", access_entity=DagAccessEntity.VERSION, user=user
+ )
+
+ return DagBundleCollectionResponse(
+ dag_bundles=[
+ DagBundleResponse(
+ name=bundle.name,
+ active=bundle.active,
+ version=bundle.version,
+ last_refreshed=bundle.last_refreshed,
Review Comment:
This can report an obsolete Git SHA as current after a Git bundle is replaced by a non-versioned bundle under the same name. Bundle sync reuses the row, and `update_bundle_state(version=None)` preserves the old version while advancing `last_refreshed`. We need to reconcile that stored state before exposing it as authoritative; otherwise this shows an old SHA with a fresh timestamp instead of “Not versioned.”
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "jayachandrakasarla (via GitHub)" <gi...@apache.org>.
jayachandrakasarla commented on PR #72909:
URL: https://github.com/apache/airflow/pull/72909#issuecomment-5631379513
It would be great to have a button in this UI to run the `airflow dags reserialize` command. I’d be happy to raise a separate pull request to integrate this feature into this page if this is something we’d like to include in the Airflow UI.
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #72909:
URL: https://github.com/apache/airflow/pull/72909#discussion_r3988223136
##########
providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py:
##########
@@ -186,6 +186,14 @@
if hasattr(MenuItem, "DEADLINES"):
_MAP_MENU_ITEM_TO_FAB_RESOURCE_TYPE[MenuItem.DEADLINES] = RESOURCE_DAG_RUN
+# Dag bundle visibility rides on Dag access -- a bundle is shown to a user who can read a Dag that
+# came from it -- so the menu entry is gated the same way the Dags entry is. Without a mapping the
+# lookup below falls back to the raw enum value as a resource name, and unlike "Jobs" or "Providers"
+# (whose enum values happen to equal an existing FAB resource) no "Dag Bundles" resource exists, so
+# the entry would silently vanish for every FAB deployment.
+if hasattr(MenuItem, "DAG_BUNDLES"):
+ _MAP_MENU_ITEM_TO_FAB_RESOURCE_TYPE[MenuItem.DAG_BUNDLES] = RESOURCE_DAG
Review Comment:
sounds good.
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #72909:
URL: https://github.com/apache/airflow/pull/72909#discussion_r4007960426
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_bundles.py:
##########
@@ -0,0 +1,214 @@
+# 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, Annotated
+
+from fastapi import Depends
+from sqlalchemy import and_, func, select
+
+from airflow.api_fastapi.auth.managers.models.resource_details import AccessView, DagAccessEntity
+from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.dag_bundles import (
+ DagBundleCollectionResponse,
+ DagBundleResponse,
+)
+from airflow.api_fastapi.core_api.security import (
+ AuthManagerDep,
+ GetUserDep,
+ ReadableDagBundlesFilterDep,
+ requires_access_dag,
+)
+from airflow.configuration import conf
+from airflow.models import DagModel
+from airflow.models.dagbundle import DagBundleModel
+from airflow.models.errors import ParseImportError
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from sqlalchemy.orm import Session
+
+ from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager
+ from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
+
+dag_bundles_router = AirflowRouter(tags=["Dag Bundle"], prefix="/dagBundles")
+
+
+def _import_error_counts(
+ *,
+ bundle_names: Sequence[str],
+ readable_dag_ids: set[str],
+ auth_manager: BaseAuthManager,
+ user: BaseUser,
+ session: Session,
+) -> dict[str, int] | None:
+ """
+ Count the import errors per bundle that this caller is allowed to know about.
+
+ Reproduces the two-part authorization of ``GET /importErrors`` rather than counting every row
+ for the bundle: an error in a file the caller can read no Dag in stays hidden, and an error in
+ a file with no registered Dag needs the admin-by-default ``IMPORT_ERRORS_ALL``, since the
+ file's existence would otherwise leak. Returns ``None`` when the caller may not read import
+ errors at all.
+ """
+ if not auth_manager.authorize_view(access_view=AccessView.IMPORT_ERRORS, user=user):
+ return None
+ if not bundle_names:
+ return {}
+
+ # Files -- keyed ``(relative_fileloc, bundle_name)`` -- in which the caller can read a Dag.
+ readable_files = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(
+ DagModel.dag_id.in_(readable_dag_ids),
+ DagModel.bundle_name.in_(bundle_names),
+ )
+ .distinct()
+ .subquery()
+ )
+ counts: dict[str, int] = {
+ bundle_name: count
+ for bundle_name, count in session.execute(
+ select(ParseImportError.bundle_name, func.count())
+ .join(
+ readable_files,
+ and_(
+ ParseImportError.filename == readable_files.c.relative_fileloc,
+ ParseImportError.bundle_name == readable_files.c.bundle_name,
+ ),
+ )
+ .group_by(ParseImportError.bundle_name)
+ ).all()
+ if bundle_name is not None
+ }
+
+ # Errors for files that never registered a Dag, added only where the caller holds the
+ # admin-by-default view for that bundle's team. Narrowed to this page's bundles, unlike the
+ # equivalent in ``import_error.py``: this endpoint polls, so an unbounded ``SELECT DISTINCT``
+ # over ``dag`` every few seconds would be a real cost on a large deployment.
+ files_with_any_dags = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(DagModel.bundle_name.in_(bundle_names))
+ .distinct()
+ .subquery()
+ )
+ unregistered = session.execute(
+ select(ParseImportError.bundle_name, func.count())
+ .outerjoin(
+ files_with_any_dags,
+ and_(
+ ParseImportError.filename == files_with_any_dags.c.relative_fileloc,
+ ParseImportError.bundle_name == files_with_any_dags.c.bundle_name,
+ ),
+ )
+ .where(
+ files_with_any_dags.c.relative_fileloc.is_(None),
+ ParseImportError.bundle_name.in_(bundle_names),
+ )
+ .group_by(ParseImportError.bundle_name)
+ ).all()
+ if unregistered:
+ team_names = DagBundleModel.get_team_names(
+ [bundle_name for bundle_name, _ in unregistered if bundle_name], session=session
+ )
+ for bundle_name, count in unregistered:
+ if bundle_name is None:
+ continue
+ if auth_manager.authorize_view(
+ access_view=AccessView.IMPORT_ERRORS_ALL,
+ user=user,
+ team_name=team_names.get(bundle_name),
+ ):
+ counts[bundle_name] = counts.get(bundle_name, 0) + count
+
+ return counts
+
+
+@dag_bundles_router.get("", dependencies=[Depends(requires_access_dag(method="GET"))])
+def get_dag_bundles(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["name", "version", "last_refreshed", "active"], DagBundleModel).dynamic_depends()),
+ ],
+ readable_dag_bundles_filter: ReadableDagBundlesFilterDep,
+ auth_manager: AuthManagerDep,
+ session: SessionDep,
+ user: GetUserDep,
+) -> DagBundleCollectionResponse:
+ """
+ List the Dag bundles Airflow knows about.
+
+ Reports the version of each bundle Airflow currently holds and when a Dag processor last
+ refreshed it, so whoever deployed the code can tell whether it has been picked up yet.
+
+ A bundle is visible to a user who can read at least one Dag recorded against it, so one from
+ which no Dag has ever parsed successfully is not listed at all.
+ """
+ bundles_select, total_entries = paginated_select(
+ statement=select(DagBundleModel),
+ filters=[readable_dag_bundles_filter],
+ order_by=order_by,
+ offset=offset,
+ limit=limit,
+ session=session,
+ )
+ bundles = session.scalars(bundles_select).all()
+ bundle_names = [bundle.name for bundle in bundles]
+
+ # ``DagBundleModel.team_name`` walks a lazily loaded relationship, so read them in one query.
+ team_names = (
+ DagBundleModel.get_team_names(bundle_names, session=session)
+ if bundle_names and conf.getboolean("core", "multi_team")
+ else {}
+ )
+
+ import_error_counts = _import_error_counts(
+ bundle_names=bundle_names,
+ readable_dag_ids=readable_dag_bundles_filter.value or set(),
+ auth_manager=auth_manager,
+ user=user,
+ session=session,
+ )
+
+ # A rendered bundle url is otherwise only reachable through ``GET /dags/{dag_id}/dagVersions``,
+ # which additionally requires Dag *version* read, so withhold it from a role without that.
+ show_bundle_url = auth_manager.is_authorized_dag(
+ method="GET", access_entity=DagAccessEntity.VERSION, user=user
+ )
+
+ return DagBundleCollectionResponse(
+ dag_bundles=[
+ DagBundleResponse(
+ name=bundle.name,
+ active=bundle.active,
+ version=bundle.version,
+ last_refreshed=bundle.last_refreshed,
Review Comment:
Confirmed the mechanism. `update_bundle_state` only writes `version` when it is not `None`, so a Git bundle replaced by a non-versioned bundle under the same name keeps the old SHA while `last_refreshed` advances.
I would rather fix that where the value is written than reconcile it in this response. The stored row is wrong either way, and anything else reading `dag_bundle.version` sees the same stale SHA, so patching it here would leave the inconsistency in place and hide it only on this page. The real fix is for the sync path to clear `version` when a bundle's classpath stops supporting versioning, which is a Dag processor behaviour change rather than an API one.
Happy to take it as a follow-up, but say if you think this should not land without it. To be clear about what this PR does and does not do: it makes a pre-existing inconsistency visible, it does not introduce 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
Re: [PR] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #72909:
URL: https://github.com/apache/airflow/pull/72909#discussion_r4007966739
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_bundles.py:
##########
@@ -0,0 +1,214 @@
+# 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, Annotated
+
+from fastapi import Depends
+from sqlalchemy import and_, func, select
+
+from airflow.api_fastapi.auth.managers.models.resource_details import AccessView, DagAccessEntity
+from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.dag_bundles import (
+ DagBundleCollectionResponse,
+ DagBundleResponse,
+)
+from airflow.api_fastapi.core_api.security import (
+ AuthManagerDep,
+ GetUserDep,
+ ReadableDagBundlesFilterDep,
+ requires_access_dag,
+)
+from airflow.configuration import conf
+from airflow.models import DagModel
+from airflow.models.dagbundle import DagBundleModel
+from airflow.models.errors import ParseImportError
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from sqlalchemy.orm import Session
+
+ from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager
+ from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
+
+dag_bundles_router = AirflowRouter(tags=["Dag Bundle"], prefix="/dagBundles")
+
+
+def _import_error_counts(
+ *,
+ bundle_names: Sequence[str],
+ readable_dag_ids: set[str],
+ auth_manager: BaseAuthManager,
+ user: BaseUser,
+ session: Session,
+) -> dict[str, int] | None:
+ """
+ Count the import errors per bundle that this caller is allowed to know about.
+
+ Reproduces the two-part authorization of ``GET /importErrors`` rather than counting every row
+ for the bundle: an error in a file the caller can read no Dag in stays hidden, and an error in
+ a file with no registered Dag needs the admin-by-default ``IMPORT_ERRORS_ALL``, since the
+ file's existence would otherwise leak. Returns ``None`` when the caller may not read import
+ errors at all.
+ """
+ if not auth_manager.authorize_view(access_view=AccessView.IMPORT_ERRORS, user=user):
+ return None
+ if not bundle_names:
+ return {}
+
+ # Files -- keyed ``(relative_fileloc, bundle_name)`` -- in which the caller can read a Dag.
+ readable_files = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(
+ DagModel.dag_id.in_(readable_dag_ids),
+ DagModel.bundle_name.in_(bundle_names),
+ )
+ .distinct()
+ .subquery()
+ )
+ counts: dict[str, int] = {
+ bundle_name: count
+ for bundle_name, count in session.execute(
+ select(ParseImportError.bundle_name, func.count())
+ .join(
+ readable_files,
+ and_(
+ ParseImportError.filename == readable_files.c.relative_fileloc,
+ ParseImportError.bundle_name == readable_files.c.bundle_name,
+ ),
+ )
+ .group_by(ParseImportError.bundle_name)
+ ).all()
+ if bundle_name is not None
+ }
+
+ # Errors for files that never registered a Dag, added only where the caller holds the
+ # admin-by-default view for that bundle's team. Narrowed to this page's bundles, unlike the
+ # equivalent in ``import_error.py``: this endpoint polls, so an unbounded ``SELECT DISTINCT``
+ # over ``dag`` every few seconds would be a real cost on a large deployment.
+ files_with_any_dags = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
+ .where(DagModel.bundle_name.in_(bundle_names))
Review Comment:
Agreed, and it is a bit worse than the two lookups you name. `DagModel.bundle_name` is a plain `ForeignKey` with no index, so it is unindexed on Postgres, though MySQL auto-creates one for FK columns. `import_error` has no index on `bundle_name` or `filename` at all.
The larger cost is upstream of the query you are pointing at. `get_authorized_dag_ids` materialises every `dag_id` on every request, and that set is then embedded as an `IN` list twice more, once in the bundle filter subquery and once in `readable_files`. On a large deployment one idle browser tab drives a full `dag` join plus two large `IN` clauses every ten seconds. `GET /dags` does the same work, but it is not on a self-refreshing timer, so this is the first endpoint where the cost becomes continuous background load rather than per-navigation.
Follow-up sounds right to me: add the indexes, and skip computing unregistered-error counts for bundles whose errors the caller cannot read. Flag it if you would rather they block this PR.
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #72909:
URL: https://github.com/apache/airflow/pull/72909#discussion_r4007954104
##########
airflow-core/docs/administration-and-deployment/dag-bundles.rst:
##########
@@ -226,6 +226,39 @@ Starting Airflow 3.0.2 git is pre installed in the base image. However, if you a
ENV GIT_PYTHON_REFRESH=quiet
+Monitoring Dag bundles
+----------------------
+
+The **Dag Bundles** page under *Browse* shows, for each bundle, the version Airflow currently holds,
+when a Dag processor last refreshed it, its import-error count, and whether it is still configured
+-- so you can tell whether a commit has been picked up without reading the Dag source. The page
+refreshes itself while open, and the same data is available at ``GET /api/v2/dagBundles`` for a
+deployment pipeline to poll.
+
+``version`` is whatever the bundle reports; for a Git Dag bundle it is the commit SHA of the tracking
+ref. Bundles that do not support versioning, such as ``LocalDagBundle``, report ``null``.
+
+``last_refreshed`` is the last *successful* refresh and advances whether or not the version changed.
+The cadence is the bundle's :ref:`config:dag_processor__refresh_interval`, checked every
+:ref:`config:dag_processor__bundle_refresh_check_interval`.
+
+Some caveats:
+
+* A failing refresh looks like one that is not due, since a refresh that raises is logged and leaves
+ ``last_refreshed`` at its last success. Check the Dag processor logs, and
+ ``/api/v2/jobs?job_type=DagProcessorJob`` for a live processor.
+* A new version does not mean the new Dags are running, because bundles are refreshed before their
+ Dags are parsed -- read the import-error count alongside it.
Review Comment:
You are right, and the timeout leg is the one I had missed. `_kill_timed_out_processors` sets `import_errors=1` on the in-memory `DagFileStat` and never writes a `ParseImportError` row, so a file that hangs past `dag_file_processor_timeout` reads as `import_error_count: 0` while parsing is genuinely broken.
Reworded in 5c3248b. The section now says outright that the version and the import-error count together cannot confirm a particular commit parsed and should not be read as a deployment-health check, and names all three reasons: errors are not recorded against a version, a timed-out parse writes no row at all, and the count only covers errors the reader is allowed to see. It adds that zero therefore means "nothing I can show you" rather than "this commit is healthy", and points at `/api/v2/dags/{dag_id}/dagVersions` for actually tying a running Dag to a commit.
--
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] Show Dag bundle versions and refresh times in the UI [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #72909:
URL: https://github.com/apache/airflow/pull/72909#discussion_r4020249923
##########
airflow-core/src/airflow/api_fastapi/common/types.py:
##########
@@ -120,6 +120,7 @@ class MenuItem(Enum):
CONFIG = "Config"
CONNECTIONS = "Connections"
DAGS = "Dags"
+ DAG_BUNDLES = "Dag Bundles"
Review Comment:
Good catch, this was a real gap. Added in 8e53572.
`MenuItem.DAG_BUNDLES` now goes into `TEAM_MENU_ITEMS` rather than `TEAM_ADMIN_MENU_ITEMS`, since bundle visibility rides on Dag read: a team Viewer who can already reach the bundles through the API should be able to find the page. `TEAM_ADMIN_MENU_ITEMS` is derived from the base set, so both roles pick it up. Guarded with `hasattr(MenuItem, "DAG_BUNDLES")` so the provider still imports against Airflow versions predating the item, matching how the FAB mapping is guarded.
The generated grants are covered: `test_create_team_command` now asserts `["Assets", "Dag Bundles", "Dags", "Docs"]` on the `MenuAccess-team-a` permission and the eight-item list on `MenuAccess-Admin-team-a`, on both the `_ensure_scope_permission` and `_attach_policy_to_scope_permission` calls. The SuperAdmin grant already built its list from `sorted(MenuItem, ...)`, so it picked the item up without a change.
--
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