Add user profile update endpoints (self-update and admin update)

This commit is contained in:
2026-01-28 17:26:04 +11:00
parent 00230f9994
commit aede5167d8

View File

@@ -5,11 +5,11 @@ from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.core.database import get_db from app.core.database import get_db
from app.core.security import verify_password, create_access_token, decode_access_token from app.core.security import verify_password, create_access_token, decode_access_token, get_password_hash
from app.core.config import settings from app.core.config import settings
from app.models.user import User from app.models.user import User
from app.schemas.auth import Token from app.schemas.auth import Token
from app.schemas.user import UserCreate, UserResponse from app.schemas.user import UserCreate, UserResponse, UserUpdate, UserAdminUpdate
router = APIRouter() router = APIRouter()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
@@ -36,6 +36,15 @@ def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(
return user return user
def get_current_admin_user(current_user: User = Depends(get_current_user)) -> User:
"""Get current user and verify they are an admin."""
if not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
return current_user
@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED) @router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register(user_data: UserCreate, db: Session = Depends(get_db)): async def register(user_data: UserCreate, db: Session = Depends(get_db)):
"""Register a new user.""" """Register a new user."""
@@ -54,11 +63,12 @@ async def register(user_data: UserCreate, db: Session = Depends(get_db)):
) )
# Create new user # Create new user
from app.core.security import get_password_hash
db_user = User( db_user = User(
username=user_data.username, username=user_data.username,
email=user_data.email, email=user_data.email,
full_name=user_data.full_name, full_name=user_data.full_name,
discord_id=user_data.discord_id,
profile_picture=user_data.profile_picture,
hashed_password=get_password_hash(user_data.password), hashed_password=get_password_hash(user_data.password),
is_active=True, is_active=True,
is_admin=False is_admin=False
@@ -100,3 +110,77 @@ async def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session =
async def get_current_user_info(current_user: User = Depends(get_current_user)): async def get_current_user_info(current_user: User = Depends(get_current_user)):
"""Get current user information.""" """Get current user information."""
return current_user return current_user
@router.put("/me", response_model=UserResponse)
async def update_current_user(
user_update: UserUpdate,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""Update current user's own profile."""
update_data = user_update.model_dump(exclude_unset=True)
# Hash password if provided
if "password" in update_data and update_data["password"]:
update_data["hashed_password"] = get_password_hash(update_data.pop("password"))
# Check email uniqueness if being updated
if "email" in update_data:
existing_user = db.query(User).filter(
User.email == update_data["email"],
User.id != current_user.id
).first()
if existing_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already in use"
)
# Update user fields
for field, value in update_data.items():
setattr(current_user, field, value)
db.commit()
db.refresh(current_user)
return current_user
@router.put("/users/{user_id}", response_model=UserResponse)
async def update_user_admin(
user_id: int,
user_update: UserAdminUpdate,
db: Session = Depends(get_db),
admin_user: User = Depends(get_current_admin_user)
):
"""Admin endpoint to update any user."""
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
update_data = user_update.model_dump(exclude_unset=True)
# Hash password if provided
if "password" in update_data and update_data["password"]:
update_data["hashed_password"] = get_password_hash(update_data.pop("password"))
# Check email uniqueness if being updated
if "email" in update_data:
existing_user = db.query(User).filter(
User.email == update_data["email"],
User.id != user.id
).first()
if existing_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already in use"
)
# Update user fields
for field, value in update_data.items():
setattr(user, field, value)
db.commit()
db.refresh(user)
return user