Migrate Flask based role_and_permission_endpoint APIs to Fastapi

providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/datamodels/roles.py

@@ -67,3 +67,10 @@ class RoleCollectionResponse(BaseModel):
 
     roles: list[RoleResponse]
     total_entries: int
+
+
+class PermissionCollectionResponse(BaseModel):
+    """Outgoing representation of a paginated collection of permissions."""
+
+    permissions: list[ActionResource] = Field(default_factory=list, serialization_alias="actions")
+    total_entries: int

providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/openapi/v2-fab-auth-manager-generated.yaml

@@ -398,6 +398,62 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
+  /auth/fab/v1/permissions:
+    get:
+      tags:
+      - FabAuthManager
+      summary: Get Permissions
+      description: List all action-resource (permission) pairs.
+      operationId: get_permissions
+      security:
+      - OAuth2PasswordBearer: []
+      - HTTPBearer: []
+      parameters:
+      - name: limit
+        in: query
+        required: false
+        schema:
+          type: integer
+          default: 100
+          title: Limit
+      - name: offset
+        in: query
+        required: false
+        schema:
+          type: integer
+          default: 0
+          title: Offset
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/PermissionCollectionResponse'
+        '401':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Unauthorized
+        '403':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Forbidden
+        '500':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Internal Server Error
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
   /auth/fab/v1/users:
     post:
       tags:
@@ -512,6 +568,21 @@ components:
       - access_token
       title: LoginResponse
       description: API Token serializer for responses.
+    PermissionCollectionResponse:
+      properties:
+        actions:
+          items:
+            $ref: '#/components/schemas/ActionResource'
+          type: array
+          title: Actions
+        total_entries:
+          type: integer
+          title: Total Entries
+      type: object
+      required:
+      - total_entries
+      title: PermissionCollectionResponse
+      description: Outgoing representation of a paginated collection of permissions.
     Resource:
       properties:
         name:

providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/routes/roles.py

@@ -23,6 +23,7 @@
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
 from airflow.providers.fab.auth_manager.api_fastapi.datamodels.roles import (
+    PermissionCollectionResponse,
     RoleBody,
     RoleCollectionResponse,
     RoleResponse,
@@ -135,3 +136,21 @@ def patch_role(
     """Update an existing role."""
     with get_application_builder():
         return FABAuthManagerRoles.patch_role(name=name, body=body, update_mask=update_mask)
+
+
+@roles_router.get(
+    "/permissions",
+    response_model=PermissionCollectionResponse,
+    responses=create_openapi_http_exception_doc(
+        [
+            status.HTTP_401_UNAUTHORIZED,
+            status.HTTP_403_FORBIDDEN,
+            status.HTTP_500_INTERNAL_SERVER_ERROR,
+        ]
+    ),
+    dependencies=[Depends(requires_fab_custom_view("GET", permissions.RESOURCE_ROLE))],
+)
+def get_permissions(limit: int = Query(100), offset: int = Query(0)):
+    """List all action-resource (permission) pairs."""
+    with get_application_builder():
+        return FABAuthManagerRoles.get_permissions(limit=limit, offset=offset)

providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/services/roles.py

@@ -20,14 +20,19 @@
 
 from fastapi import HTTPException, status
 from sqlalchemy import func, select
+from sqlalchemy.orm import joinedload
 
 from airflow.providers.fab.auth_manager.api_fastapi.datamodels.roles import (
+    Action as ActionModel,
+    ActionResource,
+    PermissionCollectionResponse,
+    Resource as ResourceModel,
     RoleBody,
     RoleCollectionResponse,
     RoleResponse,
 )
 from airflow.providers.fab.auth_manager.api_fastapi.sorting import build_ordering
-from airflow.providers.fab.auth_manager.models import Role
+from airflow.providers.fab.auth_manager.models import Permission, Role
 from airflow.providers.fab.www.utils import get_fab_auth_manager
 
 if TYPE_CHECKING:
@@ -156,3 +161,25 @@ def patch_role(cls, body: RoleBody, name: str, update_mask: str | None = None) -
         if new_name and new_name != existing.name:
             security_manager.update_role(role_id=existing.id, name=new_name)
         return RoleResponse.model_validate(update_data)
+
+    @classmethod
+    def get_permissions(cls, *, limit: int, offset: int) -> PermissionCollectionResponse:
+        security_manager = get_fab_auth_manager().security_manager
+        session = security_manager.session
+        total_entries = session.scalars(select(func.count(Permission.id))).one()
+        query = (
+            select(Permission)
+            .options(joinedload(Permission.action), joinedload(Permission.resource))
+            .offset(offset)
+            .limit(limit)
+        )
+        permissions = session.scalars(query).all()
+        return PermissionCollectionResponse(
+            permissions=[
+                ActionResource(
+                    action=ActionModel(name=p.action.name), resource=ResourceModel(name=p.resource.name)
+                )
+                for p in permissions
+            ],
+            total_entries=total_entries,
+        )

providers/fab/tests/unit/fab/auth_manager/api_fastapi/datamodels/test_roles.py

@@ -24,6 +24,7 @@
 from airflow.providers.fab.auth_manager.api_fastapi.datamodels.roles import (
     Action,
     ActionResource,
+    PermissionCollectionResponse,
     Resource,
     RoleBody,
     RoleCollectionResponse,
@@ -128,3 +129,37 @@ def test_rolecollection_model_validate_from_objects(self):
     def test_rolecollection_missing_total_entries_raises(self):
         with pytest.raises(ValidationError):
             RoleCollectionResponse.model_validate({"roles": []})
+
+    def test_permission_collection_response_valid(self):
+        ar = ActionResource(
+            action=Action(name="can_read"),
+            resource=Resource(name="DAG"),
+        )
+        resp = PermissionCollectionResponse(
+            permissions=[ar],
+            total_entries=1,
+        )
+        dumped = resp.model_dump()
+        assert dumped["total_entries"] == 1
+        assert isinstance(dumped["permissions"], list)
+        assert dumped["permissions"][0]["action"]["name"] == "can_read"
+        assert dumped["permissions"][0]["resource"]["name"] == "DAG"
+
+    def test_permission_collection_response_model_validate_from_objects(self):
+        obj = types.SimpleNamespace(
+            permissions=[
+                types.SimpleNamespace(
+                    action=types.SimpleNamespace(name="can_read"),
+                    resource=types.SimpleNamespace(name="DAG"),
+                )
+            ],
+            total_entries=1,
+        )
+        resp = PermissionCollectionResponse.model_validate(obj)
+        assert resp.total_entries == 1
+        assert len(resp.permissions) == 1
+        assert resp.permissions[0].action.name == "can_read"
+
+    def test_permission_collection_missing_total_entries_raises(self):
+        with pytest.raises(ValidationError):
+            PermissionCollectionResponse.model_validate({"permissions": []})

providers/fab/tests/unit/fab/auth_manager/api_fastapi/routes/test_roles.py

@@ -24,6 +24,10 @@
 from fastapi import HTTPException, status
 
 from airflow.providers.fab.auth_manager.api_fastapi.datamodels.roles import (
+    Action,
+    ActionResource,
+    PermissionCollectionResponse,
+    Resource,
     RoleCollectionResponse,
     RoleResponse,
 )
@@ -552,3 +556,46 @@ def test_path_role_unknown_update_mask(
             )
             assert resp.status_code == 400
             mock_roles.patch_role.assert_called_once_with(name="roleA", body=ANY, update_mask="unknown_field")
+
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.routes.roles.FABAuthManagerRoles")
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.security.get_auth_manager")
+    @patch(
+        "airflow.providers.fab.auth_manager.api_fastapi.routes.roles.get_application_builder",
+        return_value=_noop_cm(),
+    )
+    def test_get_permissions_success(
+        self, mock_get_application_builder, mock_get_auth_manager, mock_permissions, test_client, as_user
+    ):
+        mgr = MagicMock()
+        mgr.is_authorized_custom_view.return_value = True
+        mock_get_auth_manager.return_value = mgr
+
+        dummy = PermissionCollectionResponse(
+            permissions=[ActionResource(action=Action(name="can_read"), resource=Resource(name="DAG"))],
+            total_entries=1,
+        )
+        mock_permissions.get_permissions.return_value = dummy
+
+        with as_user():
+            resp = test_client.get("/fab/v1/permissions")
+            assert resp.status_code == 200
+            assert resp.json() == dummy.model_dump(by_alias=True)
+            mock_permissions.get_permissions.assert_called_once_with(limit=100, offset=0)
+
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.routes.roles.FABAuthManagerRoles")
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.security.get_auth_manager")
+    @patch(
+        "airflow.providers.fab.auth_manager.api_fastapi.routes.roles.get_application_builder",
+        return_value=_noop_cm(),
+    )
+    def test_get_permissions_forbidden(
+        self, mock_get_application_builder, mock_get_auth_manager, mock_permissions, test_client, as_user
+    ):
+        mgr = MagicMock()
+        mgr.is_authorized_custom_view.return_value = False
+        mock_get_auth_manager.return_value = mgr
+
+        with as_user():
+            resp = test_client.get("/fab/v1/permissions")
+            assert resp.status_code == 403
+            mock_permissions.get_permissions.assert_not_called()

providers/fab/tests/unit/fab/auth_manager/api_fastapi/services/test_roles.py

@@ -24,7 +24,15 @@
 from fastapi import HTTPException
 from sqlalchemy import column
 
-from airflow.providers.fab.auth_manager.api_fastapi.services.roles import FABAuthManagerRoles
+from airflow.providers.fab.auth_manager.api_fastapi.datamodels.roles import (
+    Action,
+    ActionResource,
+    PermissionCollectionResponse,
+    Resource,
+)
+from airflow.providers.fab.auth_manager.api_fastapi.services.roles import (
+    FABAuthManagerRoles,
+)
 
 
 @pytest.fixture
@@ -361,3 +369,70 @@ def test_patch_role_not_found(self, get_fab_auth_manager, fab_auth_manager, secu
         with pytest.raises(HTTPException) as ex:
             FABAuthManagerRoles.patch_role(body=body, name="viewer")
         assert ex.value.status_code == 404
+
+    def test_get_permissions_success(self, get_fab_auth_manager):
+        session = MagicMock()
+        perm_obj = types.SimpleNamespace(
+            action=types.SimpleNamespace(name="can_read"),
+            resource=types.SimpleNamespace(name="DAG"),
+        )
+        session.scalars.side_effect = [
+            types.SimpleNamespace(one=lambda: 1),
+            types.SimpleNamespace(all=lambda: [perm_obj]),
+        ]
+        fab_auth_manager = MagicMock()
+        fab_auth_manager.security_manager = MagicMock(session=session)
+        get_fab_auth_manager.return_value = fab_auth_manager
+
+        out = FABAuthManagerRoles.get_permissions(limit=10, offset=0)
+        assert isinstance(out, PermissionCollectionResponse)
+        assert out.total_entries == 1
+        assert len(out.permissions) == 1
+        assert out.permissions[0] == ActionResource(
+            action=Action(name="can_read"), resource=Resource(name="DAG")
+        )
+
+    def test_get_permissions_empty(self, get_fab_auth_manager):
+        session = MagicMock()
+        session.scalars.side_effect = [
+            types.SimpleNamespace(one=lambda: 0),
+            types.SimpleNamespace(all=lambda: []),
+        ]
+        fab_auth_manager = MagicMock()
+        fab_auth_manager.security_manager = MagicMock(session=session)
+        get_fab_auth_manager.return_value = fab_auth_manager
+
+        out = FABAuthManagerRoles.get_permissions(limit=10, offset=0)
+        assert out.total_entries == 0
+        assert out.permissions == []
+
+    def test_get_permissions_with_multiple(self, get_fab_auth_manager):
+        session = MagicMock()
+        perm_objs = [
+            types.SimpleNamespace(
+                action=types.SimpleNamespace(name="can_read"),
+                resource=types.SimpleNamespace(name="DAG"),
+            ),
+            types.SimpleNamespace(
+                action=types.SimpleNamespace(name="can_edit"),
+                resource=types.SimpleNamespace(name="DAG"),
+            ),
+        ]
+        session.scalars.side_effect = [
+            types.SimpleNamespace(one=lambda: 2),
+            types.SimpleNamespace(all=lambda: perm_objs),
+        ]
+        fab_auth_manager = MagicMock()
+        fab_auth_manager.security_manager = MagicMock(session=session)
+        get_fab_auth_manager.return_value = fab_auth_manager
+
+        out = FABAuthManagerRoles.get_permissions(limit=10, offset=0)
+        assert isinstance(out, PermissionCollectionResponse)
+        assert out.total_entries == 2
+        assert len(out.permissions) == 2
+        assert out.permissions[0] == ActionResource(
+            action=Action(name="can_read"), resource=Resource(name="DAG")
+        )
+        assert out.permissions[1] == ActionResource(
+            action=Action(name="can_edit"), resource=Resource(name="DAG")
+        )