from typing import Any from fastapi import APIRouter, Body, Depends, HTTPException, status from sqlalchemy.orm import Session from app import schemas from app.api import deps from app.models.user import User from app.services import user as user_service router = APIRouter() @router.get("/me", response_model=schemas.User) def read_user_me( current_user: User = Depends(deps.get_current_active_user), ) -> Any: """ Get current user. """ return current_user @router.put("/me", response_model=schemas.User) def update_user_me( *, db: Session = Depends(deps.get_db), full_name: str = Body(None), password: str = Body(None), current_user: User = Depends(deps.get_current_active_user), ) -> Any: """ Update current user. """ current_user_data = schemas.UserUpdate( full_name=full_name or current_user.full_name, password=password, ) user = user_service.update(db, db_obj=current_user, obj_in=current_user_data) return user @router.get("/{user_id}", response_model=schemas.User) def read_user_by_id( user_id: int, current_user: User = Depends(deps.get_current_active_user), db: Session = Depends(deps.get_db), ) -> Any: """ Get a specific user by id. """ user = user_service.get_by_id(db, id=user_id) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found", ) # Only allow superusers to access other users' data if user.id != current_user.id and not user_service.is_superuser(current_user): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions", ) return user