fat-catTW · Sep 8, 2026 · Sep 9, 2026 · Sep 9, 2026 · Oct 1, 2026 · Oct 1, 2026
diff --git a/airflow-core/docs/migrations-ref.rst b/airflow-core/docs/migrations-ref.rst
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``c71065a1b14f`` (head) | ``9cffe90fa0c0`` | ``3.5.0``         | Add session_id and registration_id to Job.                   |
+| ``a4f3c8d19e72`` (head) | ``c71065a1b14f`` | ``3.5.0``         | Allow bundle priority parsing requests.                      |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``c71065a1b14f``        | ``9cffe90fa0c0`` | ``3.5.0``         | Add session_id and registration_id to Job.                   |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``9cffe90fa0c0``        | ``a5d7b9c13e40`` | ``3.4.0``         | Add commit_allows_update to connection_test_request.         |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

diff --git a/airflow-core/docs/security/api_permissions_ref.rst b/airflow-core/docs/security/api_permissions_ref.rst
@@ -194,6 +194,10 @@ source code so it stays up to date as endpoints are added or changed.
      - ``/api/v2/dagBundles/{bundle_name}/files``
      - ``DAG``
      - ``GET``
+   * - ``POST``
+     - ``/api/v2/dagBundles/{bundle_name}/refresh``
+     - ``DAG``
+     - ``PUT``
    * - ``GET``
      - ``/api/v2/dagSources/{dag_id}``
      - ``DAG.CODE``

@@ -102,3 +102,9 @@ class DagBundleFileCollectionResponse(BaseModel):
 
     dag_bundle_files: Iterable[DagBundleFileResponse]
     total_entries: int
+
+
+class DagBundleRefreshResponse(BaseModel):
+    """Response acknowledging a Dag bundle refresh request."""
+
+    bundle_name: str
@@ -3601,6 +3601,54 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
+  /api/v2/dagBundles/{bundle_name}/refresh:
+    post:
+      tags:
+      - Dag Bundle
+      summary: Refresh Dag Bundle
+      description: Request an asynchronous Dag bundle refresh.
+      operationId: refresh_dag_bundle
+      security:
+      - OAuth2PasswordBearer: []
+      - HTTPBearer: []
+      parameters:
+      - name: bundle_name
+        in: path
+        required: true
+        schema:
+          type: string
+          title: Bundle Name
+      responses:
+        '202':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/DagBundleRefreshResponse'
+        '401':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Unauthorized
+        '403':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Forbidden
+        '404':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Not Found
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
   /api/v2/dagStats:
     get:
       tags:
@@ -15702,6 +15750,16 @@ components:
       - import_error_count
       title: DagBundleFileResponse
       description: A file in a Dag bundle, as the Dag processor last saw it.
+    DagBundleRefreshResponse:
+      properties:
+        bundle_name:
+          type: string
+          title: Bundle Name
+      type: object
+      required:
+      - bundle_name
+      title: DagBundleRefreshResponse
+      description: Response acknowledging a Dag bundle refresh request.
     DagBundleResponse:
       properties:
         name:

@@ -21,6 +21,7 @@
 
 from fastapi import Depends, HTTPException, status
 from sqlalchemy import and_, case, false, func, select
+from sqlalchemy.exc import IntegrityError
 
 from airflow.api_fastapi.auth.managers.models.resource_details import AccessView, DagAccessEntity
 from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
@@ -31,6 +32,7 @@
     DagBundleDetailResponse,
     DagBundleFileCollectionResponse,
     DagBundleFileResponse,
+    DagBundleRefreshResponse,
     DagBundleResponse,
 )
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
@@ -40,9 +42,12 @@
     PermittedDagBundleFilter,
     ReadableDagBundlesFilterDep,
     requires_access_dag,
+    requires_access_dag_bundle,
 )
+from airflow.api_fastapi.logging.decorators import action_logging
 from airflow.configuration import conf
 from airflow.models import DagModel
+from airflow.models.dagbag import DagPriorityParsingRequest
 from airflow.models.dagbundle import DagBundleModel
 from airflow.models.errors import ParseImportError
 
@@ -435,3 +440,26 @@ def get_dag_bundle_files(
         dag_bundle_files=[DagBundleFileResponse(**row) for row in ordered[start:end]],
         total_entries=len(ordered),
     )
+
+
+@dag_bundles_router.post(
+    "/{bundle_name}/refresh",
+    status_code=status.HTTP_202_ACCEPTED,
+    responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    dependencies=[Depends(action_logging()), Depends(requires_access_dag_bundle(method="PUT"))],
+)
+def refresh_dag_bundle(bundle_name: str, session: SessionDep) -> DagBundleRefreshResponse:
+    """Request an asynchronous Dag bundle refresh."""
+    try:
+        with session.begin_nested():
+            session.add(
+                DagPriorityParsingRequest(
+                    bundle_name=bundle_name,
+                    relative_fileloc=None,
+                )
+            )
+            session.flush()
+    except IntegrityError:
+        # The deterministic request key means an equivalent refresh is already pending.
+        pass
+    return DagBundleRefreshResponse(bundle_name=bundle_name)
@@ -228,6 +228,30 @@ def inner(
     return inner
 
 
+def requires_access_dag_bundle(method: ResourceMethod) -> Callable[[str, BaseUser, Session], None]:
+    """Authorize a Dag bundle action against its owning team."""
+
+    def inner(bundle_name: str, user: GetUserDep, session: SessionDep) -> None:
+        bundle = session.scalar(
+            select(DagBundleModel).where(
+                DagBundleModel.name == bundle_name,
+                DagBundleModel.active.is_(True),
+            )
+        )
+        if bundle is None:
+            raise HTTPException(status.HTTP_404_NOT_FOUND, "Dag bundle not found")
+
+        _requires_access(
+            is_authorized_callback=lambda: get_auth_manager().is_authorized_dag(
+                method=method,
+                details=DagDetails(id=None, team_name=bundle.team_name),
+                user=user,
+            )
+        )
+
+    return inner
+
+
 def requires_access_dag_from_file_token(
     method: ResourceMethod,
 ) -> Callable[[str, Request, BaseUser, Session], None]:

@@ -33,6 +33,7 @@
 from airflow.api_fastapi.core_api.security import GetUserDep
 from airflow.configuration import conf
 from airflow.models import Connection, Log, Pool, Variable
+from airflow.models.dagbundle import DagBundleModel
 from airflow.models.task_coordinates import build_coordinate_filters
 from airflow.models.taskinstance import TaskInstance
 from airflow.models.team import find_invalid_team_names
@@ -208,6 +209,8 @@ def _resolve_team_name(params: dict, *, dag_id: str | None, session: Session) ->
         return None
     if not conf.getboolean("core", "multi_team"):
         return None
+    if (bundle_name := params.get("bundle_name")) is not None:
+        return DagBundleModel.get_team_name(bundle_name, session=session)
     for param, (team_column, resource_column) in _TEAM_SCOPED_RESOURCES.items():
         if (resource_id := params.get(param)) is not None:
             return session.scalar(select(team_column).where(resource_column == resource_id))

@@ -697,11 +697,16 @@ def _claim_priority_files(self, *, session: Session = NEW_SESSION) -> list[DagFi
         )
         for request in requests:
             bundle = bundles[request.bundle_name]
-            files.append(
-                DagFileInfo(
-                    rel_path=Path(request.relative_fileloc), bundle_name=bundle.name, bundle_path=bundle.path
+            if request.relative_fileloc is None:
+                self.request_bundle_refresh(bundle.name)
+            else:
+                files.append(
+                    DagFileInfo(
+                        rel_path=Path(request.relative_fileloc),
+                        bundle_name=bundle.name,
+                        bundle_path=bundle.path,
+                    )
                 )
-            )
             session.delete(request)
         return files
 

@@ -0,0 +1,62 @@
+# 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.
+
+"""
+Allow bundle priority parsing requests.
+
+Revision ID: a4f3c8d19e72
+Revises: c71065a1b14f
+Create Date: 2026-09-08 00:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+from airflow.migrations.utils import disable_sqlite_fkeys
+
+# revision identifiers, used by Alembic.
+revision = "a4f3c8d19e72"
+down_revision = "c71065a1b14f"
+branch_labels = None
+depends_on = None
+airflow_version = "3.5.0"
+
+
+def upgrade():
+    """Allow priority parsing requests to target an entire Dag bundle."""
+    with disable_sqlite_fkeys(op):
+        with op.batch_alter_table("dag_priority_parsing_request", schema=None) as batch_op:
+            batch_op.alter_column(
+                "relative_fileloc",
+                existing_type=sa.String(length=2000),
+                nullable=True,
+            )
+
+
+def downgrade():
+    """Require priority parsing requests to target a Dag file."""
+    with disable_sqlite_fkeys(op):
+        op.execute("DELETE FROM dag_priority_parsing_request WHERE relative_fileloc IS NULL")
+        with op.batch_alter_table("dag_priority_parsing_request", schema=None) as batch_op:
+            batch_op.alter_column(
+                "relative_fileloc",
+                existing_type=sa.String(length=2000),
+                nullable=False,
+            )
@@ -296,7 +296,7 @@ def generate_md5_hash(context):
 
 
 class DagPriorityParsingRequest(Base):
-    """Model to store the dag parsing requests that will be prioritized when parsing files."""
+    """Model to store priority parsing requests for a Dag file or an entire bundle."""
 
     __tablename__ = "dag_priority_parsing_request"
 
@@ -308,13 +308,13 @@ class DagPriorityParsingRequest(Base):
     )
 
     bundle_name: Mapped[str] = mapped_column(StringID(), nullable=False)
-    # The location of the file containing the DAG object
+    # None targets the entire bundle; otherwise this is the location of the file containing the Dag object.
     # Note: Do not depend on fileloc pointing to a file; in the case of a
     # packaged DAG, it will point to the subpath of the DAG within the
     # associated zip.
-    relative_fileloc: Mapped[str] = mapped_column(String(2000), nullable=False)
+    relative_fileloc: Mapped[str | None] = mapped_column(String(2000), nullable=True)
 
-    def __init__(self, bundle_name: str, relative_fileloc: str) -> None:
+    def __init__(self, bundle_name: str, relative_fileloc: str | None) -> None:
         super().__init__()
         self.bundle_name = bundle_name
         self.relative_fileloc = relative_fileloc

@@ -1291,6 +1291,7 @@ export type DagRunServiceClearDagRunMutationResult = Awaited<ReturnType<typeof D
 export type DagRunServiceGetListDagRunsBatchMutationResult = Awaited<ReturnType<typeof DagRunService.getListDagRunsBatch>>;
 export type DagRunServiceClearDagRunsMutationResult = Awaited<ReturnType<typeof DagRunService.clearDagRuns>>;
 export type DagRunServiceClearDagRunPartitionsMutationResult = Awaited<ReturnType<typeof DagRunService.clearDagRunPartitions>>;
+export type DagBundleServiceRefreshDagBundleMutationResult = Awaited<ReturnType<typeof DagBundleService.refreshDagBundle>>;
 export type DagServiceFavoriteDagMutationResult = Awaited<ReturnType<typeof DagService.favoriteDag>>;
 export type DagServiceUnfavoriteDagMutationResult = Awaited<ReturnType<typeof DagService.unfavoriteDag>>;
 export type TaskInstanceServiceGetTaskInstancesBatchMutationResult = Awaited<ReturnType<typeof TaskInstanceService.getTaskInstancesBatch>>;

@@ -2716,6 +2716,19 @@ export const useDagRunServiceClearDagRunPartitions = <TData = Common.DagRunServi
   requestBody: ClearPartitionsBody;
 }, TContext>({ mutationFn: ({ dagId, requestBody }) => DagRunService.clearDagRunPartitions({ dagId, requestBody }) as unknown as Promise<TData>, ...options });
 /**
+* Refresh Dag Bundle
+* Request an asynchronous Dag bundle refresh.
+* @param data The data for the request.
+* @param data.bundleName
+* @returns DagBundleRefreshResponse Successful Response
+* @throws ApiError
+*/
+export const useDagBundleServiceRefreshDagBundle = <TData = Common.DagBundleServiceRefreshDagBundleMutationResult, TError = unknown, TContext = unknown>(options?: Omit<UseMutationOptions<TData, TError, {
+  bundleName: string;
+}, TContext>, "mutationFn">) => useMutation<TData, TError, {
+  bundleName: string;
+}, TContext>({ mutationFn: ({ bundleName }) => DagBundleService.refreshDagBundle({ bundleName }) as unknown as Promise<TData>, ...options });
+/**
 * Favorite Dag
 * Mark the Dag as favorite.
 * @param data The data for the request.

@@ -4926,6 +4926,19 @@ export const $DagBundleFileResponse = {
     description: 'A file in a Dag bundle, as the Dag processor last saw it.'
 } as const;
 
+export const $DagBundleRefreshResponse = {
+    properties: {
+        bundle_name: {
+            type: 'string',
+            title: 'Bundle Name'
+        }
+    },
+    type: 'object',
+    required: ['bundle_name'],
+    title: 'DagBundleRefreshResponse',
+    description: 'Response acknowledging a Dag bundle refresh request.'
+} as const;
+
 export const $DagBundleResponse = {
     properties: {
         name: {