RBAC in the API

backend/app/api/endpoints/auth.py

@@ -1,12 +1,13 @@
 from datetime import timedelta
+from typing import List
 from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.orm import Session
-from app.api.deps import get_db
+from app.api.deps import get_db, get_current_admin_user, get_current_read_user
 from app.core.config import settings
 from app.core.security import create_access_token
 from app.crud.crud_user import user as crud_user
-from app.schemas.ppr import Token
+from app.schemas.ppr import Token, UserCreate, UserUpdate, User
 
 router = APIRouter()
 
@@ -26,16 +27,63 @@ async def login_for_access_token(
             detail="Incorrect username or password",
             headers={"WWW-Authenticate": "Bearer"},
         )
-    
+
     access_token_expires = timedelta(minutes=settings.access_token_expire_minutes)
     access_token = create_access_token(
         subject=user.username, expires_delta=access_token_expires
     )
-    
+
     return {"access_token": access_token, "token_type": "bearer"}
 
 
-@router.post("/test-token")
-async def test_token(current_user = Depends(get_db)):
+@router.post("/test-token", response_model=User)
+async def test_token(current_user = Depends(get_current_read_user)):
     """Test access token"""
-    return current_user
+    return current_user
+
+
+@router.get("/users", response_model=List[User])
+async def list_users(
+    db: Session = Depends(get_db),
+    skip: int = 0,
+    limit: int = 100,
+    current_user = Depends(get_current_admin_user)
+):
+    """List all users (admin only)"""
+    users = crud_user.get_multi(db, skip=skip, limit=limit)
+    return users
+
+
+@router.post("/users", response_model=User)
+async def create_user(
+    user_in: UserCreate,
+    db: Session = Depends(get_db),
+    current_user = Depends(get_current_admin_user)
+):
+    """Create a new user (admin only)"""
+    user = crud_user.get_by_username(db, username=user_in.username)
+    if user:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Username already registered"
+        )
+    user = crud_user.create(db, obj_in=user_in)
+    return user
+
+
+@router.put("/users/{user_id}", response_model=User)
+async def update_user(
+    user_id: int,
+    user_in: UserUpdate,
+    db: Session = Depends(get_db),
+    current_user = Depends(get_current_admin_user)
+):
+    """Update a user (admin only)"""
+    user = crud_user.get(db, user_id=user_id)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="User not found"
+        )
+    user = crud_user.update(db, db_obj=user, obj_in=user_in)
+    return user