Coverage for api/users/controller.py: 92.96%
71 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-03 15:30 +0000
1import redis
2from fastapi import APIRouter, Depends, HTTPException, Path, Query, Request, Response
3from sqlalchemy.ext.asyncio import AsyncSession
5from core.dependencies import get_db
6from core.permissions import Permission
7from core.rbac import require_permission
8from core.redis import get_redis
9from core.security import verify_token
10from utils.custom_exception import AuthorizationException, ConflictException, NotFoundException
11from utils.response import APIResponse, common_responses, parse_responses
13from .schema import (
14 PasswordReset,
15 UserCreate,
16 UserDelete,
17 UserDeleteBatchResponse,
18 UserPagination,
19 UserResponse,
20 UserSortBy,
21 UserUpdate,
22 user_delete_failed_response_example,
23 user_delete_partial_response_example,
24 user_delete_success_response_example,
25)
26from .services import create_user, delete_users, get_all_users, reset_user_password, update_user
28router = APIRouter(tags=["Users"])
31@router.get(
32 "",
33 response_model=APIResponse[UserPagination],
34 summary="Get all users",
35 responses=parse_responses(
36 {200: ("Successfully retrieved users", UserPagination)}, common_responses
37 ),
38)
39@require_permission([Permission.VIEW_USERS, Permission.MANAGE_USERS])
40async def get_users(
41 request: Request,
42 token: dict = Depends(verify_token),
43 db: AsyncSession = Depends(get_db),
44 keyword: str | None = Query(None, description="Keyword to search for users"),
45 status: str | None = Query(
46 None,
47 description="Filter user status (multiple values separated by commas, example: true,false)",
48 ),
49 role: str | None = Query(
50 None,
51 description=(
52 "Filter user role (multiple values separated by commas, example: admin,manager)"
53 ),
54 ),
55 page: int = Query(1, ge=1, description="Page number"),
56 per_page: int = Query(10, ge=1, le=100, description="Number of users per page"),
57 sort_by: UserSortBy | None = Query(None, description="Sort by field"),
58 desc: bool = Query(False, description="Sort order"),
59):
60 try:
61 data = await get_all_users(
62 db=db,
63 keyword=keyword,
64 status=status,
65 role=role,
66 page=page,
67 per_page=per_page,
68 sort_by=sort_by.value if sort_by else None,
69 desc=desc,
70 )
71 return APIResponse(code=200, message="Successfully retrieved users", data=data)
72 except Exception:
73 raise HTTPException(status_code=500)
76@router.post(
77 "",
78 response_model=APIResponse[UserResponse],
79 response_model_exclude_none=True,
80 summary="Create new user",
81 responses=parse_responses({200: ("User created successfully", UserResponse)}, common_responses),
82)
83@require_permission([Permission.MANAGE_USERS])
84async def create_user_api(
85 user_data: UserCreate,
86 request: Request,
87 token: dict = Depends(verify_token),
88 db: AsyncSession = Depends(get_db),
89):
90 """Create a new user account"""
91 try:
92 user = await create_user(db, user_data, actor_user_id=token["sub"])
93 return APIResponse(code=200, message="User created successfully", data=user)
94 except AuthorizationException as e:
95 raise HTTPException(status_code=403, detail=e.message)
96 except Exception as e:
97 if "Email already exists" in str(e):
98 raise HTTPException(status_code=409, detail="Email already exists")
99 raise HTTPException(status_code=500)
102@router.put(
103 "/{user_id}",
104 response_model=APIResponse[UserResponse],
105 response_model_exclude_none=True,
106 summary="Update user info",
107 responses=parse_responses({200: ("User updated successfully", UserResponse)}, common_responses),
108)
109@require_permission([Permission.MANAGE_USERS])
110async def update_user_api(
111 request: Request = None,
112 token: dict = Depends(verify_token),
113 user_id: str = Path(..., description="User ID"),
114 user_data: UserUpdate = None,
115 db: AsyncSession = Depends(get_db),
116):
117 """Update user information"""
118 try:
119 user = await update_user(db, user_id, user_data, actor_user_id=token["sub"])
120 return APIResponse(code=200, message="User updated successfully", data=user)
121 except NotFoundException:
122 raise HTTPException(status_code=404, detail="User not found")
123 except ConflictException:
124 raise HTTPException(status_code=409, detail="Email already exists")
125 except AuthorizationException as e:
126 raise HTTPException(status_code=403, detail=e.message)
127 except Exception:
128 raise HTTPException(status_code=500)
131@router.delete(
132 "",
133 response_model=APIResponse[UserDeleteBatchResponse],
134 response_model_exclude_none=True,
135 summary="Delete users",
136 responses=parse_responses(
137 {
138 200: (
139 "All users deleted successfully",
140 UserDeleteBatchResponse,
141 user_delete_success_response_example,
142 ),
143 207: (
144 "Users deleted with partial success",
145 UserDeleteBatchResponse,
146 user_delete_partial_response_example,
147 ),
148 400: (
149 "All users failed to delete",
150 UserDeleteBatchResponse,
151 user_delete_failed_response_example,
152 ),
153 },
154 common_responses,
155 ),
156)
157@require_permission([Permission.MANAGE_USERS])
158async def delete_users_api(
159 delete_data: UserDelete,
160 request: Request,
161 token: dict = Depends(verify_token),
162 db: AsyncSession = Depends(get_db),
163 redis_client: redis.Redis = Depends(get_redis),
164):
165 """Delete multiple users"""
166 try:
167 batch_result = await delete_users(db, redis_client, delete_data.user_ids, token)
169 # Determine response code based on results
170 if batch_result.failed_count == 0:
171 # All successful
172 return APIResponse(
173 code=200, message="All users deleted successfully", data=batch_result
174 )
175 elif batch_result.success_count == 0:
176 # All failed - return 400 status code
177 response = APIResponse(
178 code=400, message="All users failed to delete", data=batch_result
179 )
180 return Response(
181 content=response.model_dump_json(), status_code=400, media_type="application/json"
182 )
183 else:
184 # Partial success - return 207 status code
185 response = APIResponse(
186 code=207, message="Users deleted with partial success", data=batch_result
187 )
188 return Response(
189 content=response.model_dump_json(), status_code=207, media_type="application/json"
190 )
191 except Exception:
192 raise HTTPException(status_code=500)
195@router.post(
196 "/{user_id}/reset-password",
197 response_model=APIResponse[dict],
198 response_model_exclude_none=True,
199 summary="Reset user password",
200 responses=parse_responses({200: ("Password reset successfully", dict)}, common_responses),
201)
202@require_permission([Permission.MANAGE_USERS])
203async def reset_user_password_api(
204 user_id: str = Path(..., description="User ID"),
205 password_data: PasswordReset = None,
206 request: Request = None,
207 token: dict = Depends(verify_token),
208 db: AsyncSession = Depends(get_db),
209 redis_client: redis.Redis = Depends(get_redis),
210):
211 """Reset user password and logout all devices"""
212 try:
213 await reset_user_password(db, redis_client, user_id, password_data.new_password)
214 return APIResponse(
215 code=200, message="Password reset successfully and all devices logged out"
216 )
217 except NotFoundException:
218 raise HTTPException(status_code=404, detail="User not found")
219 except Exception:
220 raise HTTPException(status_code=500)