Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions app/api/routes/admin/admins.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import uuid
from typing import Annotated

from fastapi import APIRouter, Request, status
from fastapi import APIRouter, Query, Request, status

from app.api.decorators import audit_unexpected_failure
from app.api.deps import CurrentSuperAdmin, SessionDep
Expand All @@ -15,7 +16,7 @@
RootTransferRequest,
)
from app.schemas.msg import Message
from app.schemas.user import Language
from app.schemas.user import Language, SystemRole
from app.schemas.user_activity import ActivityType, ResourceType
from app.services.admin.admin_service import (
confirm_root_transfer_service,
Expand All @@ -42,9 +43,12 @@ async def list_admins_endpoint(
_request: Request,
_admin: CurrentSuperAdmin,
session: SessionDep,
skip: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=200)] = 50,
role: SystemRole | None = None,
) -> AdminListResponse:
"""List every admin-tier account with its permissions (superadmin only)."""
return await list_admins_service(session=session)
"""List admin-tier accounts with pagination and an optional role filter (superadmin only)."""
return await list_admins_service(session=session, skip=skip, limit=limit, role=role)


@router.get("/permissions", response_model=PermissionCatalogResponse)
Expand Down
29 changes: 22 additions & 7 deletions app/repositories/admin/user.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,14 +66,29 @@ async def list_users_admin(
return users, total


async def list_admins(session: AsyncSession) -> Sequence[User]:
"""Return all admin and superadmin accounts, newest first."""
stmt = (
select(User)
.where(User.role.in_([SystemRole.ADMIN.value, SystemRole.SUPERADMIN.value]))
.order_by(User.created_at.desc())
async def list_admins(
session: AsyncSession,
*,
skip: int = 0,
limit: int = 50,
role: SystemRole | None = None,
) -> tuple[Sequence[User], int]:
"""Return a paginated page of admin-tier accounts plus the total count."""
base_stmt = select(User).where(
User.role.in_([SystemRole.ADMIN.value, SystemRole.SUPERADMIN.value])
)
return (await session.execute(stmt)).scalars().all()
if role is not None:
base_stmt = base_stmt.where(User.role == role.value)

count_stmt = base_stmt.with_only_columns(
func.count(), maintain_column_froms=True
).order_by(None)
total = (await session.execute(count_stmt)).scalar_one()

rows_stmt = base_stmt.order_by(User.created_at.desc()).offset(skip).limit(limit)
admins = (await session.execute(rows_stmt)).scalars().all()

return admins, total


async def superadmin_exists(session: AsyncSession) -> bool:
Expand Down
5 changes: 4 additions & 1 deletion app/schemas/admin.py
Original file line number Diff line number Diff line change
Expand Up @@ -194,13 +194,16 @@ class AdminListItem(BaseModel):
is_active: bool
is_root_superadmin: bool = False
permissions: list[Permission] = Field(default_factory=list)
created_at: datetime


class AdminListResponse(BaseModel):
"""Listing of every admin-tier account."""
"""Paginated listing of admin-tier accounts."""

data: list[AdminListItem]
total: int
skip: int
limit: int


class AdminCreate(BaseModel):
Expand Down
15 changes: 11 additions & 4 deletions app/services/admin/admin_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ def _to_list_item(user: User, permissions: list[Permission]) -> AdminListItem:
is_active=user.is_active,
is_root_superadmin=user.is_root_superadmin,
permissions=permissions,
created_at=user.created_at,
)


Expand Down Expand Up @@ -104,9 +105,15 @@ def get_permission_catalog_service() -> PermissionCatalogResponse:
return PermissionCatalogResponse(permissions=list(Permission))


async def list_admins_service(session: AsyncSession) -> AdminListResponse:
"""List every admin-tier account with the permissions it holds."""
admins = await list_admins(session)
async def list_admins_service(
session: AsyncSession,
*,
skip: int = 0,
limit: int = 50,
role: SystemRole | None = None,
) -> AdminListResponse:
"""List a paginated page of admin-tier accounts with their permissions."""
admins, total = await list_admins(session, skip=skip, limit=limit, role=role)
permissions_map = await get_permissions_for_users(
session, [admin.id for admin in admins]
)
Expand All @@ -116,7 +123,7 @@ async def list_admins_service(session: AsyncSession) -> AdminListResponse:
)
for admin in admins
]
return AdminListResponse(data=data, total=len(data))
return AdminListResponse(data=data, total=total, skip=skip, limit=limit)


async def create_admin_service(
Expand Down
42 changes: 42 additions & 0 deletions app/tests/admin/test_admins.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,48 @@ async def test_list_admins_reports_superadmin_with_all_permissions(
assert superadmin_row["is_root_superadmin"] is True


@pytest.mark.asyncio
async def test_list_admins_paginates_and_reports_total(superadmin_client: AsyncClient):
"""``skip``/``limit`` slice the listing while ``total`` counts every admin."""
for i in range(3):
await register_and_verify(superadmin_client, f"pageadmin{i}@test.com")
await promote_to_admin(f"pageadmin{i}@test.com")

full = await superadmin_client.get("/admin/admins")
expected_total = full.json()["total"]
assert expected_total >= 4 # 3 admins + the seeded superadmin

first = await superadmin_client.get("/admin/admins", params={"skip": 0, "limit": 2})
assert first.status_code == 200
body = first.json()
assert body["skip"] == 0
assert body["limit"] == 2
assert len(body["data"]) == 2
assert body["total"] == expected_total

second = await superadmin_client.get(
"/admin/admins", params={"skip": 2, "limit": 2}
)
first_ids = {row["id"] for row in body["data"]}
second_ids = {row["id"] for row in second.json()["data"]}
assert first_ids.isdisjoint(second_ids)


@pytest.mark.asyncio
async def test_list_admins_role_filter(superadmin_client: AsyncClient):
"""The optional role filter narrows the listing to a single tier."""
await register_and_verify(superadmin_client, "roleadmin@test.com")
await promote_to_admin("roleadmin@test.com")

response = await superadmin_client.get(
"/admin/admins", params={"role": SystemRole.SUPERADMIN.value}
)
assert response.status_code == 200
rows = response.json()["data"]
assert rows
assert all(row["role"] == SystemRole.SUPERADMIN.value for row in rows)


# --- Create admin -----------------------------------------------------------


Expand Down
Loading