Coverage for api/users/controller.py: 97.06%

1import redis 

2from typing import Optional 

3from core.redis import get_redis 

4from core.dependencies import get_db 

5from core.security import verify_token 

6from core.permissions import Permission 

7from core.rbac import require_permission 

8from sqlalchemy.ext.asyncio import AsyncSession 

9from utils.response import APIResponse, parse_responses, common_responses 

10from fastapi import APIRouter, Depends, HTTPException, Query, Request, Path, Response 

11from .services import get_all_users, create_user, update_user, delete_users, reset_user_password 

12from .schema import ( 

13 UserPagination, UserSortBy, UserCreate, UserUpdate, UserDelete, PasswordReset, UserResponse, 

14 UserDeleteBatchResponse, user_delete_success_response_example, user_delete_partial_response_example, 

15 user_delete_failed_response_example 

16) 

17from utils.custom_exception import NotFoundException, ConflictException 

18 

19router = APIRouter(tags=["Users"]) 

20 

21@router.get( 

22 "", 

23 response_model=APIResponse[UserPagination], 

24 summary="Get all users", 

25 responses=parse_responses({ 

26 200: ("Successfully retrieved users", UserPagination) 

27 }, common_responses) 

28) 

29@require_permission([Permission.VIEW_USERS, Permission.MANAGE_USERS]) 

30async def get_users( 

31 request: Request, 

32 token: dict = Depends(verify_token), 

33 db: AsyncSession = Depends(get_db), 

34 keyword: Optional[str] = Query(None, description="Keyword to search for users"), 

35 status: Optional[str] = Query(None, description="Filter user status (multiple values separated by commas, example: true,false)"), 

36 role: Optional[str] = Query(None, description="Filter user role (multiple values separated by commas, example: admin,manager)"), 

37 page: int = Query(1, ge=1, description="Page number"), 

38 per_page: int = Query(10, ge=1, le=100, description="Number of users per page"), 

39 sort_by: Optional[UserSortBy] = Query(None, description="Sort by field"), 

40 desc: bool = Query(False, description="Sort order") 

41): 

42 try: 

43 data = await get_all_users( 

44 db=db, 

45 keyword=keyword, 

46 status=status, 

47 role=role, 

48 page=page, 

49 per_page=per_page, 

50 sort_by=sort_by.value if sort_by else None, 

51 desc=desc 

52 ) 

53 return APIResponse(code=200, message="Successfully retrieved users", data=data) 

54 except Exception: 

55 raise HTTPException(status_code=500) 

56 

57@router.post( 

58 "", 

59 response_model=APIResponse[UserResponse], 

60 response_model_exclude_none=True, 

61 summary="Create new user", 

62 responses=parse_responses({ 

63 200: ("User created successfully", UserResponse) 

64 }, common_responses) 

65) 

66@require_permission([Permission.MANAGE_USERS]) 

67async def create_user_api( 

68 user_data: UserCreate, 

69 request: Request, 

70 token: dict = Depends(verify_token), 

71 db: AsyncSession = Depends(get_db) 

72): 

73 """Create a new user account""" 

74 try: 

75 user = await create_user(db, user_data) 

76 return APIResponse(code=200, message="User created successfully", data=user) 

77 except Exception as e: 

78 if "Email already exists" in str(e): 

79 raise HTTPException(status_code=409, detail="Email already exists") 

80 raise HTTPException(status_code=500) 

81 

82@router.put( 

83 "/{user_id}", 

84 response_model=APIResponse[UserResponse], 

85 response_model_exclude_none=True, 

86 summary="Update user info", 

87 responses=parse_responses({ 

88 200: ("User updated successfully", UserResponse) 

89 }, common_responses) 

90) 

91@require_permission([Permission.MANAGE_USERS]) 

92async def update_user_api( 

93 request: Request = None, 

94 token: dict = Depends(verify_token), 

95 user_id: str = Path(..., description="User ID"), 

96 user_data: UserUpdate = None, 

97 db: AsyncSession = Depends(get_db) 

98): 

99 """Update user information""" 

100 try: 

101 user = await update_user(db, user_id, user_data) 

102 return APIResponse(code=200, message="User updated successfully", data=user) 

103 except NotFoundException: 

104 raise HTTPException(status_code=404, detail="User not found") 

105 except ConflictException: 

106 raise HTTPException(status_code=409, detail="Email already exists") 

107 except Exception: 

108 raise HTTPException(status_code=500) 

109 

110@router.delete( 

111 "", 

112 response_model=APIResponse[UserDeleteBatchResponse], 

113 response_model_exclude_none=True, 

114 summary="Delete users", 

115 responses=parse_responses({ 

116 200: ("All users deleted successfully", UserDeleteBatchResponse, user_delete_success_response_example), 

117 207: ("Users deleted with partial success", UserDeleteBatchResponse, user_delete_partial_response_example), 

118 400: ("All users failed to delete", UserDeleteBatchResponse, user_delete_failed_response_example) 

119 }, common_responses) 

120) 

121@require_permission([Permission.MANAGE_USERS]) 

122async def delete_users_api( 

123 delete_data: UserDelete, 

124 request: Request, 

125 token: dict = Depends(verify_token), 

126 db: AsyncSession = Depends(get_db), 

127 redis_client: redis.Redis = Depends(get_redis) 

128): 

129 """Delete multiple users""" 

130 try: 

131 batch_result = await delete_users(db, redis_client, delete_data.user_ids, token) 

132 

133 # Determine response code based on results 

134 if batch_result.failed_count == 0: 

135 # All successful 

136 return APIResponse( 

137 code=200, 

138 message="All users deleted successfully", 

139 data=batch_result 

140 ) 

141 elif batch_result.success_count == 0: 

142 # All failed - return 400 status code 

143 response = APIResponse( 

144 code=400, 

145 message="All users failed to delete", 

146 data=batch_result 

147 ) 

148 return Response( 

149 content=response.model_dump_json(), 

150 status_code=400, 

151 media_type="application/json" 

152 ) 

153 else: 

154 # Partial success - return 207 status code 

155 response = APIResponse( 

156 code=207, 

157 message="Users deleted with partial success", 

158 data=batch_result 

159 ) 

160 return Response( 

161 content=response.model_dump_json(), 

162 status_code=207, 

163 media_type="application/json" 

164 ) 

165 except Exception: 

166 raise HTTPException(status_code=500) 

167 

168@router.post( 

169 "/{user_id}/reset-password", 

170 response_model=APIResponse[dict], 

171 response_model_exclude_none=True, 

172 summary="Reset user password", 

173 responses=parse_responses({ 

174 200: ("Password reset successfully", dict) 

175 }, common_responses) 

176) 

177@require_permission([Permission.MANAGE_USERS]) 

178async def reset_user_password_api( 

179 user_id: str = Path(..., description="User ID"), 

180 password_data: PasswordReset = None, 

181 request: Request = None, 

182 token: dict = Depends(verify_token), 

183 db: AsyncSession = Depends(get_db), 

184 redis_client: redis.Redis = Depends(get_redis) 

185): 

186 """Reset user password and logout all devices""" 

187 try: 

188 await reset_user_password(db, redis_client, user_id, password_data.new_password) 

189 return APIResponse(code=200, message="Password reset successfully and all devices logged out") 

190 except NotFoundException: 

191 raise HTTPException(status_code=404, detail="User not found") 

192 except Exception: 

193 raise HTTPException(status_code=500)