|
| 1 | +"""Admin API routes for user role/plan management. |
| 2 | +
|
| 3 | +Only accessible by users with role='admin' or listed in VERTEX_ADMIN_USER_IDS/EMAILS. |
| 4 | +""" |
| 5 | + |
| 6 | +from __future__ import annotations |
| 7 | + |
| 8 | +import logging |
| 9 | +from typing import Any |
| 10 | + |
| 11 | +from bson import ObjectId |
| 12 | +from bson.errors import InvalidId |
| 13 | +from fastapi import APIRouter, Depends, HTTPException |
| 14 | +from pydantic import BaseModel, Field |
| 15 | + |
| 16 | +from app.api.deps import User, get_current_user |
| 17 | +from app.database.repositories.user import UserRepository |
| 18 | +from app.services.provider_routing import _is_admin |
| 19 | + |
| 20 | +logger = logging.getLogger(__name__) |
| 21 | + |
| 22 | +router = APIRouter(prefix="/admin", tags=["admin"]) |
| 23 | + |
| 24 | +VALID_ROLES = {"user", "admin"} |
| 25 | +VALID_PLANS = {"free", "premium", "pro", "enterprise"} |
| 26 | + |
| 27 | + |
| 28 | +async def require_admin(user: User = Depends(get_current_user)) -> User: |
| 29 | + """Dependency that requires the current user to be an admin.""" |
| 30 | + user_repo = UserRepository() |
| 31 | + user_doc = await user_repo.get_by_id(user.id) |
| 32 | + if not _is_admin(user_doc): |
| 33 | + raise HTTPException(status_code=403, detail="Admin access required") |
| 34 | + return user |
| 35 | + |
| 36 | + |
| 37 | +class UserRoleUpdate(BaseModel): |
| 38 | + role: str | None = Field(None, description="User role (user, admin)") |
| 39 | + plan: str | None = Field( |
| 40 | + None, description="User plan (free, premium, pro, enterprise)" |
| 41 | + ) |
| 42 | + |
| 43 | + |
| 44 | +class AdminUserResponse(BaseModel): |
| 45 | + id: str |
| 46 | + username: str |
| 47 | + email: str | None = None |
| 48 | + github_id: str | None = None |
| 49 | + avatar_url: str | None = None |
| 50 | + role: str = "user" |
| 51 | + plan: str = "free" |
| 52 | + created_at: str | None = None |
| 53 | + |
| 54 | + |
| 55 | +def _to_admin_response(u: dict) -> AdminUserResponse: |
| 56 | + return AdminUserResponse( |
| 57 | + id=str(u["_id"]), |
| 58 | + username=u.get("username", ""), |
| 59 | + email=u.get("email"), |
| 60 | + github_id=str(u.get("github_id", "")), |
| 61 | + avatar_url=u.get("avatar_url"), |
| 62 | + role=u.get("role", "user"), |
| 63 | + plan=u.get("plan", "free"), |
| 64 | + created_at=str(u.get("created_at", "")), |
| 65 | + ) |
| 66 | + |
| 67 | + |
| 68 | +@router.get("/users", response_model=list[AdminUserResponse]) |
| 69 | +async def list_users( |
| 70 | + _admin: User = Depends(require_admin), |
| 71 | +) -> list[AdminUserResponse]: |
| 72 | + """List all users with their roles and plans.""" |
| 73 | + user_repo = UserRepository() |
| 74 | + users = await user_repo.list(limit=500) |
| 75 | + return [_to_admin_response(u) for u in users] |
| 76 | + |
| 77 | + |
| 78 | +@router.patch("/users/{user_id}", response_model=AdminUserResponse) |
| 79 | +async def update_user_role( |
| 80 | + user_id: str, |
| 81 | + update: UserRoleUpdate, |
| 82 | + _admin: User = Depends(require_admin), |
| 83 | +) -> AdminUserResponse: |
| 84 | + """Update a user's role and/or plan.""" |
| 85 | + try: |
| 86 | + ObjectId(user_id) |
| 87 | + except (InvalidId, Exception): |
| 88 | + raise HTTPException(status_code=400, detail="Invalid user ID format") |
| 89 | + |
| 90 | + update_data: dict[str, Any] = {} |
| 91 | + |
| 92 | + if update.role is not None: |
| 93 | + if update.role not in VALID_ROLES: |
| 94 | + raise HTTPException( |
| 95 | + status_code=400, |
| 96 | + detail=f"Invalid role: {update.role}. Must be one of {VALID_ROLES}", |
| 97 | + ) |
| 98 | + if update.role != "admin" and user_id == _admin.id: |
| 99 | + raise HTTPException( |
| 100 | + status_code=400, |
| 101 | + detail="Cannot demote yourself. Ask another admin.", |
| 102 | + ) |
| 103 | + update_data["role"] = update.role |
| 104 | + |
| 105 | + if update.plan is not None: |
| 106 | + if update.plan not in VALID_PLANS: |
| 107 | + raise HTTPException( |
| 108 | + status_code=400, |
| 109 | + detail=f"Invalid plan: {update.plan}. Must be one of {VALID_PLANS}", |
| 110 | + ) |
| 111 | + update_data["plan"] = update.plan |
| 112 | + |
| 113 | + if not update_data: |
| 114 | + raise HTTPException(status_code=400, detail="No fields to update") |
| 115 | + |
| 116 | + user_repo = UserRepository() |
| 117 | + updated = await user_repo.update_user(user_id, update_data) |
| 118 | + |
| 119 | + if not updated: |
| 120 | + raise HTTPException(status_code=404, detail="User not found") |
| 121 | + |
| 122 | + logger.info( |
| 123 | + "[Admin] User %s updated: %s by admin %s", user_id, update_data, _admin.id |
| 124 | + ) |
| 125 | + |
| 126 | + return _to_admin_response(updated) |
0 commit comments