Spaces:
Running
Running
| from fastapi import APIRouter, Depends, HTTPException, status, Query | |
| from sqlalchemy.orm import Session | |
| from typing import List, Optional | |
| from uuid import UUID | |
| from app.core import get_db, get_current_user, get_current_admin_user, get_password_hash | |
| from app.models import User, UserRole | |
| from app.schemas import UserCreate, UserUpdate, UserResponse | |
| router = APIRouter(prefix="/users", tags=["User Management"]) | |
| async def get_current_user_info(current_user: User = Depends(get_current_user)): | |
| """Get current user information""" | |
| return current_user | |
| async def list_users( | |
| skip: int = Query(0, ge=0), | |
| limit: int = Query(50, ge=1, le=100), | |
| db: Session = Depends(get_db), | |
| current_user: User = Depends(get_current_admin_user) | |
| ): | |
| """List all users (admin only)""" | |
| users = db.query(User).offset(skip).limit(limit).all() | |
| return users | |
| async def create_user( | |
| user_data: UserCreate, | |
| db: Session = Depends(get_db), | |
| current_user: User = Depends(get_current_admin_user) | |
| ): | |
| """Create a new user (admin only)""" | |
| # Check if username or email already exists | |
| if db.query(User).filter(User.username == user_data.username).first(): | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Username already registered" | |
| ) | |
| if db.query(User).filter(User.email == user_data.email).first(): | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Email already registered" | |
| ) | |
| user = User( | |
| email=user_data.email, | |
| username=user_data.username, | |
| hashed_password=get_password_hash(user_data.password), | |
| full_name=user_data.full_name, | |
| phone=user_data.phone, | |
| role=user_data.role | |
| ) | |
| db.add(user) | |
| db.commit() | |
| db.refresh(user) | |
| return user | |
| async def get_user( | |
| user_id: UUID, | |
| db: Session = Depends(get_db), | |
| current_user: User = Depends(get_current_admin_user) | |
| ): | |
| """Get user by ID (admin only)""" | |
| 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" | |
| ) | |
| return user | |
| async def update_user( | |
| user_id: UUID, | |
| user_data: UserUpdate, | |
| db: Session = Depends(get_db), | |
| current_user: User = Depends(get_current_admin_user) | |
| ): | |
| """Update user (admin only)""" | |
| 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_data.model_dump(exclude_unset=True) | |
| for field, value in update_data.items(): | |
| setattr(user, field, value) | |
| db.commit() | |
| db.refresh(user) | |
| return user | |
| async def delete_user( | |
| user_id: UUID, | |
| db: Session = Depends(get_db), | |
| current_user: User = Depends(get_current_admin_user) | |
| ): | |
| """Delete user (admin only)""" | |
| if str(user_id) == str(current_user.id): | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Cannot delete yourself" | |
| ) | |
| 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" | |
| ) | |
| db.delete(user) | |
| db.commit() | |
| return None | |