Group and filter Dags by folder on the Dags page

airflow-core/src/airflow/api_fastapi/common/parameters.py

@@ -1044,6 +1044,43 @@ def depends(cls, teams: list[str] = Query(default_factory=list)) -> _TeamsFilter
         return cls().set_value(teams)
 
 
+class _RelativeFilelocPrefixFilter(BaseParam[str | None]):
+    """
+    Filter Dags by the folder they live in, derived from ``relative_fileloc``.
+
+    The value is treated as a directory path relative to the bundle root (e.g.
+    ``team_a/etl``). It matches every Dag whose file lives directly in that folder
+    or in any of its subfolders, using an escaped ``LIKE 'team_a/etl/%'`` so a
+    folder name is never a substring/prefix of another (``team_a`` won't match
+    ``team_alpha``). Dags at the bundle root (no ``/`` in ``relative_fileloc``)
+    are not matched by any folder value and appear only when no folder is selected.
+    """
+
+    def to_orm(self, select: Select) -> Select:
+        if self.value is None and self.skip_none:
+            return select
+
+        if not self.value:
+            return select
+
+        directory = self.value.rstrip("/")
+        escaped = _escape_like_pattern(directory)
+        return select.where(DagModel.relative_fileloc.like(f"{escaped}/%", escape=_LIKE_ESCAPE_CHAR))
+
+    @classmethod
+    def depends(
+        cls,
+        relative_fileloc_prefix: str | None = Query(
+            default=None,
+            description=(
+                "Filter Dags by the folder (directory of ``relative_fileloc``) they live in. "
+                "Matches the given folder and all of its subfolders."
+            ),
+        ),
+    ) -> _RelativeFilelocPrefixFilter:
+        return cls().set_value(relative_fileloc_prefix)
+
+
 def _safe_parse_datetime(date_to_check: str) -> datetime:
     """
     Parse datetime and raise error for invalid dates.
@@ -1293,6 +1330,9 @@ def depends_float(
 QueryTagsFilter = Annotated[_TagsFilter, Depends(_TagsFilter.depends)]
 QueryOwnersFilter = Annotated[_OwnersFilter, Depends(_OwnersFilter.depends)]
 QueryTeamsFilter = Annotated[_TeamsFilter, Depends(_TeamsFilter.depends)]
+QueryRelativeFilelocPrefixFilter = Annotated[
+    _RelativeFilelocPrefixFilter, Depends(_RelativeFilelocPrefixFilter.depends)
+]
 
 
 class _HasAssetScheduleFilter(BaseParam[bool]):

airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py

@@ -54,3 +54,17 @@ class DAGsRunStateCountsCollectionResponse(BaseModel):
 
     dags: list[DAGRunStateCountsResponse]
     state_count_limit: int
+
+
+class DagFolderResponse(BaseModel):
+    """A distinct Dag folder (directory of ``relative_fileloc``) within a bundle."""
+
+    bundle_name: str
+    folder: str
+
+
+class DagFolderCollectionResponse(BaseModel):
+    """Collection of distinct Dag folders, each scoped to the bundle it belongs to."""
+
+    folders: list[DagFolderResponse]
+    total_entries: int

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -509,6 +509,18 @@ paths:
           - type: string
           - type: 'null'
           title: Bundle Version
+      - name: relative_fileloc_prefix
+        in: query
+        required: false
+        schema:
+          anyOf:
+          - type: string
+          - type: 'null'
+          description: Filter Dags by the folder (directory of ``relative_fileloc``)
+            they live in. Matches the given folder and all of its subfolders.
+          title: Relative Fileloc Prefix
+        description: Filter Dags by the folder (directory of ``relative_fileloc``)
+          they live in. Matches the given folder and all of its subfolders.
       - name: order_by
         in: query
         required: false
@@ -574,6 +586,43 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
+  /ui/dags/folders:
+    get:
+      tags:
+      - DAG
+      summary: Get Dag Folders
+      description: 'Get the distinct folders the readable Dags live in, scoped to
+        their bundle.
+
+
+        A folder is the directory part of a Dag''s ``relative_fileloc`` (relative
+        to its
+
+        bundle root). Because ``relative_fileloc`` is relative to each bundle, the
+        same
+
+        path can exist in several bundles, so every folder is paired with its bundle
+
+        name to keep them apart. Dags located directly at the bundle root have no
+        folder
+
+        and are not represented here. The result powers the folder navigation tree
+        in
+
+        the UI, which reconstructs the hierarchy by splitting each path on ``/`` and
+
+        groups it under its bundle when more than one bundle is present.'
+      operationId: get_dag_folders
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/DagFolderCollectionResponse'
+      security:
+      - OAuth2PasswordBearer: []
+      - HTTPBearer: []
   /ui/dags/{dag_id}/latest_run:
     get:
       tags:
@@ -2706,6 +2755,38 @@ components:
       - state_count_limit
       title: DAGsRunStateCountsCollectionResponse
       description: Collection of per-Dag DagRun-state counts for the Dag list page.
+    DagFolderCollectionResponse:
+      properties:
+        folders:
+          items:
+            $ref: '#/components/schemas/DagFolderResponse'
+          type: array
+          title: Folders
+        total_entries:
+          type: integer
+          title: Total Entries
+      type: object
+      required:
+      - folders
+      - total_entries
+      title: DagFolderCollectionResponse
+      description: Collection of distinct Dag folders, each scoped to the bundle it
+        belongs to.
+    DagFolderResponse:
+      properties:
+        bundle_name:
+          type: string
+          title: Bundle Name
+        folder:
+          type: string
+          title: Folder
+      type: object
+      required:
+      - bundle_name
+      - folder
+      title: DagFolderResponse
+      description: A distinct Dag folder (directory of ``relative_fileloc``) within
+        a bundle.
     DagRunState:
       type: string
       enum:

airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+from pathlib import PurePosixPath
 from typing import Annotated
 
 from fastapi import Depends, HTTPException, Query, status
@@ -50,6 +51,7 @@
     QueryOwnersFilter,
     QueryPausedFilter,
     QueryPendingActionsFilter,
+    QueryRelativeFilelocPrefixFilter,
     QueryTagsFilter,
     QueryTeamsFilter,
     SortParam,
@@ -59,6 +61,8 @@
 from airflow.api_fastapi.core_api.datamodels.dags import DAG_ALIAS_MAPPING, DAGResponse
 from airflow.api_fastapi.core_api.datamodels.ui.dag_runs import DAGRunLightResponse
 from airflow.api_fastapi.core_api.datamodels.ui.dags import (
+    DagFolderCollectionResponse,
+    DagFolderResponse,
     DAGRunStateCountsResponse,
     DAGsRunStateCountsCollectionResponse,
     DAGWithLatestDagRunsCollectionResponse,
@@ -113,6 +117,7 @@ def get_dags(
     dag_run_state: QueryAnyDagRunStateFilter,
     bundle_name: QueryBundleNameFilter,
     bundle_version: QueryBundleVersionFilter,
+    relative_fileloc_prefix: QueryRelativeFilelocPrefixFilter,
     order_by: Annotated[
         SortParam,
         Depends(
@@ -169,6 +174,7 @@ def get_dags(
             readable_dags_filter,
             bundle_name,
             bundle_version,
+            relative_fileloc_prefix,
         ],
         order_by=order_by,
         offset=offset,
@@ -275,6 +281,47 @@ def get_dags(
     )
 
 
+@dags_router.get(
+    "/folders",
+    dependencies=[Depends(requires_access_dag(method="GET"))],
+    operation_id="get_dag_folders",
+)
+def get_dag_folders(
+    readable_dags_filter: ReadableDagsFilterDep,
+    session: SessionDep,
+) -> DagFolderCollectionResponse:
+    """
+    Get the distinct folders the readable Dags live in, scoped to their bundle.
+
+    A folder is the directory part of a Dag's ``relative_fileloc`` (relative to its
+    bundle root). Because ``relative_fileloc`` is relative to each bundle, the same
+    path can exist in several bundles, so every folder is paired with its bundle
+    name to keep them apart. Dags located directly at the bundle root have no folder
+    and are not represented here. The result powers the folder navigation tree in
+    the UI, which reconstructs the hierarchy by splitting each path on ``/`` and
+    groups it under its bundle when more than one bundle is present.
+    """
+    query = readable_dags_filter.to_orm(
+        select(DagModel.bundle_name, DagModel.relative_fileloc)
+        .where(DagModel.relative_fileloc.is_not(None))
+        .distinct()
+    )
+    folders: set[tuple[str, str]] = set()
+    for bundle_name, relative_fileloc in session.execute(query):
+        parent = PurePosixPath(relative_fileloc).parent
+        if str(parent) != ".":
+            folders.add((bundle_name, str(parent)))
+
+    sorted_folders = sorted(folders)
+    return DagFolderCollectionResponse(
+        folders=[
+            DagFolderResponse(bundle_name=bundle_name, folder=folder)
+            for bundle_name, folder in sorted_folders
+        ],
+        total_entries=len(sorted_folders),
+    )
+
+
 @dags_router.get(
     "/{dag_id}/latest_run",
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),

airflow-core/src/airflow/ui/openapi-gen/queries/common.ts

@@ -339,7 +339,7 @@ export const UseDagServiceGetDagTagsKeyFn = ({ limit, offset, orderBy, tagNamePa
 export type DagServiceGetDagsUiDefaultResponse = Awaited<ReturnType<typeof DagService.getDagsUi>>;
 export type DagServiceGetDagsUiQueryResult<TData = DagServiceGetDagsUiDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
 export const useDagServiceGetDagsUiKey = "DagServiceGetDagsUi";
-export const UseDagServiceGetDagsUiKeyFn = ({ assetDependency, bundleName, bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode, teams }: {
+export const UseDagServiceGetDagsUiKeyFn = ({ assetDependency, bundleName, bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners, paused, relativeFilelocPrefix, tags, tagsMatchMode, teams }: {
   assetDependency?: string;
   bundleName?: string;
   bundleVersion?: string;
@@ -361,10 +361,15 @@ export const UseDagServiceGetDagsUiKeyFn = ({ assetDependency, bundleName, bundl
   orderBy?: string[];
   owners?: string[];
   paused?: boolean;
+  relativeFilelocPrefix?: string;
   tags?: string[];
   tagsMatchMode?: "any" | "all";
   teams?: string[];
-} = {}, queryKey?: Array<unknown>) => [useDagServiceGetDagsUiKey, ...(queryKey ?? [{ assetDependency, bundleName, bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode, teams }])];
+} = {}, queryKey?: Array<unknown>) => [useDagServiceGetDagsUiKey, ...(queryKey ?? [{ assetDependency, bundleName, bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners, paused, relativeFilelocPrefix, tags, tagsMatchMode, teams }])];
+export type DagServiceGetDagFoldersDefaultResponse = Awaited<ReturnType<typeof DagService.getDagFolders>>;
+export type DagServiceGetDagFoldersQueryResult<TData = DagServiceGetDagFoldersDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
+export const useDagServiceGetDagFoldersKey = "DagServiceGetDagFolders";
+export const UseDagServiceGetDagFoldersKeyFn = (queryKey?: Array<unknown>) => [useDagServiceGetDagFoldersKey, ...(queryKey ?? [])];
 export type DagServiceGetLatestRunInfoDefaultResponse = Awaited<ReturnType<typeof DagService.getLatestRunInfo>>;
 export type DagServiceGetLatestRunInfoQueryResult<TData = DagServiceGetLatestRunInfoDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
 export const useDagServiceGetLatestRunInfoKey = "DagServiceGetLatestRunInfo";

airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts

@@ -692,6 +692,7 @@ export const ensureUseDagServiceGetDagTagsData = (queryClient: QueryClient, { li
 * @param data.dagRunState Filter Dags that have any DagRun in the given state. Only ``queued`` and ``running`` are supported.
 * @param data.bundleName
 * @param data.bundleVersion
+* @param data.relativeFilelocPrefix Filter Dags by the folder (directory of ``relative_fileloc``) they live in. Matches the given folder and all of its subfolders.
 * @param data.orderBy Attributes to order by, multi criteria sort is supported. Prefix with `-` for descending order. Supported attributes: `dag_id, dag_display_name, next_dagrun, state, start_date, last_run_state, last_run_start_date, last_run_run_after`
 * @param data.isFavorite
 * @param data.hasAssetSchedule Filter Dags with asset-based scheduling
@@ -700,7 +701,7 @@ export const ensureUseDagServiceGetDagTagsData = (queryClient: QueryClient, { li
 * @returns DAGWithLatestDagRunsCollectionResponse Successful Response
 * @throws ApiError
 */
-export const ensureUseDagServiceGetDagsUiData = (queryClient: QueryClient, { assetDependency, bundleName, bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode, teams }: {
+export const ensureUseDagServiceGetDagsUiData = (queryClient: QueryClient, { assetDependency, bundleName, bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners, paused, relativeFilelocPrefix, tags, tagsMatchMode, teams }: {
   assetDependency?: string;
   bundleName?: string;
   bundleVersion?: string;
@@ -722,10 +723,26 @@ export const ensureUseDagServiceGetDagsUiData = (queryClient: QueryClient, { ass
   orderBy?: string[];
   owners?: string[];
   paused?: boolean;
+  relativeFilelocPrefix?: string;
   tags?: string[];
   tagsMatchMode?: "any" | "all";
   teams?: string[];
-} = {}) => queryClient.ensureQueryData({ queryKey: Common.UseDagServiceGetDagsUiKeyFn({ assetDependency, bundleName, bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode, teams }), queryFn: () => DagService.getDagsUi({ assetDependency, bundleName, bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode, teams }) });
+} = {}) => queryClient.ensureQueryData({ queryKey: Common.UseDagServiceGetDagsUiKeyFn({ assetDependency, bundleName, bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners, paused, relativeFilelocPrefix, tags, tagsMatchMode, teams }), queryFn: () => DagService.getDagsUi({ assetDependency, bundleName, bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners, paused, relativeFilelocPrefix, tags, tagsMatchMode, teams }) });
+/**
+* Get Dag Folders
+* Get the distinct folders the readable Dags live in, scoped to their bundle.
+*
+* A folder is the directory part of a Dag's ``relative_fileloc`` (relative to its
+* bundle root). Because ``relative_fileloc`` is relative to each bundle, the same
+* path can exist in several bundles, so every folder is paired with its bundle
+* name to keep them apart. Dags located directly at the bundle root have no folder
+* and are not represented here. The result powers the folder navigation tree in
+* the UI, which reconstructs the hierarchy by splitting each path on ``/`` and
+* groups it under its bundle when more than one bundle is present.
+* @returns DagFolderCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const ensureUseDagServiceGetDagFoldersData = (queryClient: QueryClient) => queryClient.ensureQueryData({ queryKey: Common.UseDagServiceGetDagFoldersKeyFn(), queryFn: () => DagService.getDagFolders() });
 /**
 * Get Latest Run Info
 * Get latest run.