chat_fast/chatfast/api/admin.py

"""Administrative API routes."""

from typing import Any, Dict, Optional

from fastapi import APIRouter, Depends

from chatfast.services.auth import (
    AdminUserRequest,
    AdminUserUpdateRequest,
    UserInfo,
    admin_create_user as admin_create_user_service,
    admin_delete_user as admin_delete_user_service,
    admin_get_user as admin_get_user_service,
    admin_list_users as admin_list_users_service,
    admin_update_user as admin_update_user_service,
    require_admin,
)
from chatfast.services.chat import list_exports_admin

router = APIRouter(prefix="/api/admin", tags=["admin"])


@router.get("/users")
async def api_admin_list_users(
    keyword: Optional[str] = None,
    page: int = 0,
    page_size: int = 20,
    admin: UserInfo = Depends(require_admin),
) -> Dict[str, Any]:
    return await admin_list_users_service(keyword, page, page_size)


@router.post("/users")
async def api_admin_create_user(payload: AdminUserRequest, admin: UserInfo = Depends(require_admin)) -> Dict[str, Any]:
    return await admin_create_user_service(payload.username, payload.password)


@router.get("/users/{user_id}")
async def api_admin_get_user(user_id: int, admin: UserInfo = Depends(require_admin)) -> Dict[str, Any]:
    return await admin_get_user_service(user_id)


@router.put("/users/{user_id}")
async def api_admin_update_user(
    user_id: int,
    payload: AdminUserUpdateRequest,
    admin: UserInfo = Depends(require_admin),
) -> Dict[str, Any]:
    return await admin_update_user_service(user_id, payload)


@router.delete("/users/{user_id}")
async def api_admin_delete_user(user_id: int, admin: UserInfo = Depends(require_admin)) -> Dict[str, Any]:
    return await admin_delete_user_service(user_id)


@router.get("/exports")
async def api_admin_exports(
    keyword: Optional[str] = None,
    admin: UserInfo = Depends(require_admin),
) -> Dict[str, Any]:
    items = await list_exports_admin(keyword)
    return {"items": items}