161 lines
5 KiB
Python
161 lines
5 KiB
Python
"""
|
|
User routes for the backend API.
|
|
Implements the endpoints for user management.
|
|
"""
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Body, Path, Request, Response
|
|
from typing import List, Dict, Any, Optional
|
|
from fastapi import status
|
|
from datetime import datetime
|
|
import logging
|
|
import inspect
|
|
import importlib
|
|
import os
|
|
from pydantic import BaseModel
|
|
|
|
# Import interfaces and models
|
|
import modules.interfaces.serviceAppClass as serviceAppClass
|
|
from modules.security.auth import getCurrentUser, limiter, getCurrentUser
|
|
|
|
# Import the attribute definition and helper functions
|
|
from modules.interfaces.serviceAppModel import User, AttributeDefinition
|
|
from modules.shared.attributeUtils import getModelAttributeDefinitions, AttributeResponse
|
|
|
|
# Configure logger
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(
|
|
prefix="/api/users",
|
|
tags=["Manage Users"],
|
|
responses={404: {"description": "Not found"}}
|
|
)
|
|
|
|
@router.get("/", response_model=List[User])
|
|
@limiter.limit("30/minute")
|
|
async def get_users(
|
|
request: Request,
|
|
mandateId: Optional[str] = None,
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> List[User]:
|
|
"""Get all users in the current mandate"""
|
|
try:
|
|
appInterface = serviceAppClass.getInterface(currentUser)
|
|
# If mandateId is provided, use it, otherwise use the current user's mandate
|
|
targetMandateId = mandateId or currentUser.mandateId
|
|
# Get all users without filtering by enabled status
|
|
users = appInterface.getUsersByMandate(targetMandateId)
|
|
return users
|
|
except Exception as e:
|
|
logger.error(f"Error getting users: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to get users: {str(e)}"
|
|
)
|
|
|
|
@router.get("/{userId}", response_model=User)
|
|
@limiter.limit("30/minute")
|
|
async def get_user(
|
|
request: Request,
|
|
userId: str = Path(..., description="ID of the user"),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> User:
|
|
"""Get a specific user by ID"""
|
|
try:
|
|
appInterface = serviceAppClass.getInterface(currentUser)
|
|
# Get user without filtering by enabled status
|
|
user = appInterface.getUser(userId)
|
|
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"User with ID {userId} not found"
|
|
)
|
|
|
|
return user
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting user {userId}: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to get user: {str(e)}"
|
|
)
|
|
|
|
@router.post("", response_model=User)
|
|
@limiter.limit("10/minute")
|
|
async def create_user(
|
|
request: Request,
|
|
user_data: User = Body(...),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> User:
|
|
"""Create a new user"""
|
|
appInterface = serviceAppClass.getInterface(currentUser)
|
|
|
|
# Convert User to dict for interface
|
|
user_dict = user_data.dict()
|
|
|
|
# Create user
|
|
newUser = appInterface.createUser(user_dict)
|
|
|
|
return newUser
|
|
|
|
@router.put("/{userId}", response_model=User)
|
|
@limiter.limit("10/minute")
|
|
async def update_user(
|
|
request: Request,
|
|
userId: str = Path(..., description="ID of the user to update"),
|
|
userData: User = Body(...),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> User:
|
|
"""Update an existing user"""
|
|
appInterface = serviceAppClass.getInterface(currentUser)
|
|
|
|
# Check if the user exists
|
|
existingUser = appInterface.getUser(userId)
|
|
if not existingUser:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"User with ID {userId} not found"
|
|
)
|
|
|
|
# Convert User to dict for interface
|
|
update_data = userData.dict()
|
|
|
|
# Update user
|
|
updatedUser = appInterface.updateUser(userId, update_data)
|
|
|
|
if not updatedUser:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Error updating the user"
|
|
)
|
|
|
|
return updatedUser
|
|
|
|
@router.delete("/{userId}", response_model=Dict[str, Any])
|
|
@limiter.limit("10/minute")
|
|
async def delete_user(
|
|
request: Request,
|
|
userId: str = Path(..., description="ID of the user to delete"),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> Dict[str, Any]:
|
|
"""Delete a user"""
|
|
appInterface = serviceAppClass.getInterface(currentUser)
|
|
|
|
# Check if the user exists
|
|
existingUser = appInterface.getUser(userId)
|
|
if not existingUser:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"User with ID {userId} not found"
|
|
)
|
|
|
|
success = appInterface.deleteUser(userId)
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Error deleting the user"
|
|
)
|
|
|
|
return {"message": f"User with ID {userId} successfully deleted"}
|
|
|