Coverage for api/v1/endpoints/users.py: 45.45%

1""" 

2用户管理API端点 

3""" 

4 

5from typing import List 

6 

7from fastapi import APIRouter, Depends, HTTPException, status 

8 

9from core.middleware.auth_middleware import get_admin_user, get_current_user 

10from core.middleware.permission_middleware import require_admin_permission 

11from core.models.user import User, UserCreate, UserResponse, UserUpdate 

12from core.services.user_service import UserService 

13 

14router = APIRouter() 

15user_service = UserService() 

16 

17 

18@router.get("/users", response_model=List[UserResponse]) 

19async def get_all_users(current_user: User = Depends(get_admin_user)): 

20 """获取所有用户（仅管理员）""" 

21 users = user_service.get_all_users(current_user) 

22 if users is None: 

23 raise HTTPException( 

24 status_code=status.HTTP_403_FORBIDDEN, detail="需要管理员权限" 

25 ) 

26 return users 

27 

28 

29@router.post("/users", response_model=UserResponse) 

30async def create_user( 

31 user_data: UserCreate, current_user: User = Depends(get_admin_user) 

32): 

33 """创建用户（仅管理员）""" 

34 user = user_service.create_user(user_data, current_user) 

35 if not user: 

36 raise HTTPException( 

37 status_code=status.HTTP_400_BAD_REQUEST, 

38 detail="创建用户失败，用户名可能已存在", 

39 ) 

40 return user 

41 

42 

43@router.get("/users/{user_id}", response_model=UserResponse) 

44async def get_user(user_id: str, current_user: User = Depends(get_current_user)): 

45 """获取用户信息""" 

46 user = user_service.get_user_by_id(user_id, current_user) 

47 if not user: 

48 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在") 

49 return user 

50 

51 

52@router.put("/users/{user_id}", response_model=UserResponse) 

53async def update_user( 

54 user_id: str, user_data: UserUpdate, current_user: User = Depends(get_current_user) 

55): 

56 """更新用户信息""" 

57 user = user_service.update_user(user_id, user_data, current_user) 

58 if not user: 

59 raise HTTPException( 

60 status_code=status.HTTP_400_BAD_REQUEST, detail="更新用户失败" 

61 ) 

62 return user 

63 

64 

65@router.delete("/users/{user_id}") 

66async def delete_user(user_id: str, current_user: User = Depends(get_admin_user)): 

67 """删除用户（仅管理员）""" 

68 success = user_service.delete_user(user_id, current_user) 

69 if not success: 

70 raise HTTPException( 

71 status_code=status.HTTP_400_BAD_REQUEST, detail="删除用户失败" 

72 ) 

73 return {"message": "用户删除成功"} 

74 

75 

76@router.get("/users/{user_id}/accounts") 

77async def get_user_accounts( 

78 user_id: str, current_user: User = Depends(get_current_user) 

79): 

80 """获取用户账户信息""" 

81 accounts = user_service.get_user_accounts(user_id, current_user) 

82 if accounts is None: 

83 raise HTTPException( 

84 status_code=status.HTTP_403_FORBIDDEN, detail="无权限访问该用户账户信息" 

85 ) 

86 return accounts