173 lines
4.4 KiB
Python
173 lines
4.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from starlette import status
|
|
|
|
from app.api.dependencies import get_admin_user, get_user_service
|
|
from app.models.auth import UserResponse, MessageResponse
|
|
from app.models.types import PyObjectId
|
|
from app.models.user import UserInDB, UserCreate, UserUpdate
|
|
from app.services.user_service import UserService
|
|
|
|
router = APIRouter(tags=["users"])
|
|
|
|
|
|
@router.get("", response_model=list[UserInDB])
|
|
def list_users(
|
|
admin_user: UserInDB = Depends(get_admin_user),
|
|
user_service: UserService = Depends(get_user_service)
|
|
):
|
|
"""
|
|
List all users (admin only).
|
|
|
|
Args:
|
|
admin_user: Current admin user
|
|
user_service: User service instance
|
|
|
|
Returns:
|
|
List[UserResponse]: List of all users without sensitive data
|
|
"""
|
|
return user_service.list_users()
|
|
|
|
|
|
@router.get("/{user_id}", response_model=UserResponse)
|
|
def get_user_by_id(
|
|
user_id: PyObjectId,
|
|
admin_user: UserInDB = Depends(get_admin_user),
|
|
user_service: UserService = Depends(get_user_service)
|
|
):
|
|
"""
|
|
Get specific user by ID (admin only).
|
|
|
|
Args:
|
|
user_id: User ID to retrieve
|
|
admin_user: Current admin user
|
|
user_service: User service instance
|
|
|
|
Returns:
|
|
UserResponse: User information without sensitive data
|
|
|
|
Raises:
|
|
HTTPException: If user not found
|
|
"""
|
|
user = user_service.get_user_by_id(str(user_id))
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User not found"
|
|
)
|
|
|
|
return user
|
|
|
|
|
|
@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_user(
|
|
user_data: UserCreate,
|
|
admin_user: UserInDB = Depends(get_admin_user),
|
|
user_service: UserService = Depends(get_user_service)
|
|
):
|
|
"""
|
|
Create new user (admin only).
|
|
|
|
Args:
|
|
user_data: User creation data
|
|
admin_user: Current admin user
|
|
user_service: User service instance
|
|
|
|
Returns:
|
|
UserResponse: Created user information without sensitive data
|
|
|
|
Raises:
|
|
HTTPException: If user creation fails
|
|
"""
|
|
try:
|
|
user = user_service.create_user(user_data)
|
|
return UserResponse(
|
|
_id=user.id,
|
|
username=user.username,
|
|
email=user.email,
|
|
role=user.role,
|
|
is_active=user.is_active,
|
|
created_at=user.created_at,
|
|
updated_at=user.updated_at
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e)
|
|
)
|
|
|
|
|
|
@router.put("/{user_id}", response_model=UserResponse)
|
|
def update_user(
|
|
user_id: PyObjectId,
|
|
user_data: UserUpdate,
|
|
admin_user: UserInDB = Depends(get_admin_user),
|
|
user_service: UserService = Depends(get_user_service)
|
|
):
|
|
"""
|
|
Update existing user (admin only).
|
|
|
|
Args:
|
|
user_id: User ID to update
|
|
user_data: User update data
|
|
admin_user: Current admin user
|
|
user_service: User service instance
|
|
|
|
Returns:
|
|
UserResponse: Updated user information without sensitive data
|
|
|
|
Raises:
|
|
HTTPException: If user not found or update fails
|
|
"""
|
|
try:
|
|
user = user_service.update_user(str(user_id), user_data)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User not found"
|
|
)
|
|
|
|
return UserResponse(
|
|
_id=user.id,
|
|
username=user.username,
|
|
email=user.email,
|
|
role=user.role,
|
|
is_active=user.is_active,
|
|
created_at=user.created_at,
|
|
updated_at=user.updated_at
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e)
|
|
)
|
|
|
|
|
|
@router.delete("/{user_id}", response_model=MessageResponse)
|
|
def delete_user(
|
|
user_id: PyObjectId,
|
|
admin_user: UserInDB = Depends(get_admin_user),
|
|
user_service: UserService = Depends(get_user_service)
|
|
):
|
|
"""
|
|
Delete user by ID (admin only).
|
|
|
|
Args:
|
|
user_id: User ID to delete
|
|
admin_user: Current admin user
|
|
user_service: User service instance
|
|
|
|
Returns:
|
|
MessageResponse: Success message
|
|
|
|
Raises:
|
|
HTTPException: If user not found or deletion fails
|
|
"""
|
|
success = user_service.delete_user(str(user_id))
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User not found"
|
|
)
|
|
|
|
return MessageResponse(message="User successfully deleted")
|