Coverage for api/roles/controller.py: 98.99%
99 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-03 15:30 +0000
1from fastapi import APIRouter, Depends, HTTPException, Path, Request, Response
2from sqlalchemy.ext.asyncio import AsyncSession
4from core.dependencies import get_db
5from core.permissions import Permission
6from core.rbac import require_permission
7from core.security import verify_token
8from utils.custom_exception import (
9 AuthorizationException,
10 ConflictException,
11 NotFoundException,
12)
13from utils.response import APIResponse, common_responses, parse_responses
15from .schema import (
16 PermissionCheckResponse,
17 RoleAttributeMappingBatchResponse,
18 RoleAttributesGroupedResponse,
19 RoleAttributesMapping,
20 RoleCreate,
21 RoleResponse,
22 RolesListResponse,
23 RoleUpdate,
24 role_attributes_failed_response_example,
25 role_attributes_partial_response_example,
26 role_attributes_success_response_example,
27)
28from .services import (
29 check_user_permissions,
30 create_role,
31 delete_role,
32 get_all_roles,
33 get_role_attribute_mapping,
34 update_role,
35 update_role_attribute_mapping,
36)
38router = APIRouter(tags=["Roles"])
41@router.get(
42 "",
43 response_model=APIResponse[RolesListResponse],
44 response_model_exclude_none=True,
45 summary="Get all custom roles",
46 responses=parse_responses(
47 {200: ("Successfully retrieved roles", RolesListResponse)}, common_responses
48 ),
49)
50@require_permission([Permission.VIEW_ROLES, Permission.MANAGE_ROLES])
51async def get_roles(
52 request: Request, token: dict = Depends(verify_token), db: AsyncSession = Depends(get_db)
53):
54 """Get all custom roles"""
55 try:
56 roles = await get_all_roles(db, actor_user_id=token["sub"])
57 return APIResponse(code=200, message="Successfully retrieved roles", data=roles)
58 except Exception:
59 raise HTTPException(status_code=500)
62@router.post(
63 "",
64 response_model=APIResponse[RoleResponse],
65 response_model_exclude_none=True,
66 summary="Create new role",
67 responses=parse_responses(
68 {200: ("Role created successfully", RoleResponse), 409: ("Role name already exists", None)},
69 common_responses,
70 ),
71)
72@require_permission([Permission.MANAGE_ROLES])
73async def create_role_api(
74 role_data: RoleCreate,
75 request: Request,
76 token: dict = Depends(verify_token),
77 db: AsyncSession = Depends(get_db),
78):
79 """Create a new role"""
80 try:
81 role = await create_role(db, role_data, actor_user_id=token["sub"])
82 return APIResponse(code=200, message="Role created successfully", data=role)
83 except AuthorizationException as e:
84 raise HTTPException(status_code=403, detail=e.message)
85 except ConflictException:
86 raise HTTPException(status_code=409, detail="Role name already exists")
87 except Exception:
88 raise HTTPException(status_code=500)
91@router.put(
92 "/{role_id}",
93 response_model=APIResponse[RoleResponse],
94 response_model_exclude_none=True,
95 summary="Update role info",
96 responses=parse_responses(
97 {
98 200: ("Role updated successfully", RoleResponse),
99 404: ("Role not found", None),
100 409: ("Role name already exists", None),
101 },
102 common_responses,
103 ),
104)
105@require_permission([Permission.MANAGE_ROLES])
106async def update_role_api(
107 role_id: str = Path(..., description="Role ID"),
108 role_data: RoleUpdate = None,
109 request: Request = None,
110 token: dict = Depends(verify_token),
111 db: AsyncSession = Depends(get_db),
112):
113 """Update role information"""
114 try:
115 role = await update_role(db, role_id, role_data, actor_user_id=token["sub"])
116 return APIResponse(code=200, message="Role updated successfully", data=role)
117 except NotFoundException:
118 raise HTTPException(status_code=404, detail="Role not found")
119 except AuthorizationException as e:
120 raise HTTPException(status_code=403, detail=e.message)
121 except ConflictException:
122 raise HTTPException(status_code=409, detail="Role name already exists")
123 except Exception:
124 raise HTTPException(status_code=500)
127@router.delete(
128 "/{role_id}",
129 response_model=APIResponse[dict],
130 response_model_exclude_none=True,
131 summary="Delete role",
132 responses=parse_responses(
133 {
134 200: ("Role deleted successfully", None),
135 404: ("Role not found", None),
136 409: ("Cannot delete role that is assigned to users", None),
137 },
138 common_responses,
139 ),
140)
141@require_permission([Permission.MANAGE_ROLES])
142async def delete_role_api(
143 role_id: str = Path(..., description="Role ID"),
144 request: Request = None,
145 token: dict = Depends(verify_token),
146 db: AsyncSession = Depends(get_db),
147):
148 """Delete a role"""
149 try:
150 await delete_role(db, role_id, actor_user_id=token["sub"])
151 return APIResponse(code=200, message="Role deleted successfully")
152 except NotFoundException:
153 raise HTTPException(status_code=404, detail="Role not found")
154 except AuthorizationException as e:
155 raise HTTPException(status_code=403, detail=e.message)
156 except ConflictException:
157 raise HTTPException(status_code=409, detail="Cannot delete role that is assigned to users")
158 except Exception:
159 raise HTTPException(status_code=500)
162@router.get(
163 "/{role_id}/attributes",
164 response_model=APIResponse[RoleAttributesGroupedResponse],
165 response_model_exclude_none=True,
166 summary="Get role attributes mapping",
167 responses=parse_responses(
168 {
169 200: (
170 "Successfully retrieved role attributes mapping",
171 RoleAttributesGroupedResponse,
172 RoleAttributesGroupedResponse.get_example_response(),
173 ),
174 404: ("Role not found", None),
175 },
176 common_responses,
177 ),
178)
179@require_permission([Permission.VIEW_ROLES, Permission.MANAGE_ROLES])
180async def get_role_attribute_mapping_api(
181 role_id: str = Path(..., description="Role ID"),
182 request: Request = None,
183 token: dict = Depends(verify_token),
184 db: AsyncSession = Depends(get_db),
185):
186 """Get role attributes mapping with all available attributes"""
187 try:
188 attributes_mapping = await get_role_attribute_mapping(db, role_id)
189 return APIResponse(
190 code=200,
191 message="Successfully retrieved role attributes mapping",
192 data=attributes_mapping,
193 )
194 except NotFoundException:
195 raise HTTPException(status_code=404, detail="Role not found")
196 except AuthorizationException as e:
197 raise HTTPException(status_code=403, detail=e.message)
198 except Exception:
199 raise HTTPException(status_code=500)
202@router.put(
203 "/{role_id}/attributes",
204 response_model=APIResponse[RoleAttributeMappingBatchResponse],
205 response_model_exclude_none=True,
206 summary="Update role attributes",
207 responses=parse_responses(
208 {
209 200: (
210 "All role attributes processed successfully",
211 RoleAttributeMappingBatchResponse,
212 role_attributes_success_response_example,
213 ),
214 207: (
215 "Role attributes processed with partial success",
216 RoleAttributeMappingBatchResponse,
217 role_attributes_partial_response_example,
218 ),
219 400: (
220 "All role attributes failed to process",
221 RoleAttributeMappingBatchResponse,
222 role_attributes_failed_response_example,
223 ),
224 },
225 common_responses,
226 ),
227)
228@require_permission([Permission.MANAGE_ROLES])
229async def update_role_attribute_mapping_api(
230 role_id: str = Path(..., description="Role ID"),
231 attributes_data: RoleAttributesMapping = None,
232 request: Request = None,
233 token: dict = Depends(verify_token),
234 db: AsyncSession = Depends(get_db),
235):
236 """Update role attributes with batch processing results"""
237 try:
238 batch_result = await update_role_attribute_mapping(
239 db, role_id, attributes_data.attributes, actor_user_id=token["sub"]
240 )
242 # Determine response code based on results
243 if batch_result.failed_count == 0:
244 # All successful
245 return APIResponse(
246 code=200, message="All role attributes processed successfully", data=batch_result
247 )
248 elif batch_result.success_count == 0:
249 # All failed - return 400 status code
250 response = APIResponse(
251 code=400, message="All role attributes failed to process", data=batch_result
252 )
253 return Response(
254 content=response.model_dump_json(), status_code=400, media_type="application/json"
255 )
256 else:
257 # Partial success - return 207 status code
258 response = APIResponse(
259 code=207,
260 message="Role attributes processed with partial success",
261 data=batch_result,
262 )
263 return Response(
264 content=response.model_dump_json(), status_code=207, media_type="application/json"
265 )
266 except NotFoundException:
267 raise HTTPException(status_code=404, detail="Role not found")
268 except AuthorizationException as e:
269 raise HTTPException(status_code=403, detail=e.message)
270 except Exception:
271 raise HTTPException(status_code=500)
274@router.get(
275 "/permissions",
276 response_model=APIResponse[PermissionCheckResponse],
277 response_model_exclude_none=True,
278 summary="Get current user permissions",
279 responses=parse_responses(
280 {
281 200: (
282 "User permissions retrieved",
283 PermissionCheckResponse,
284 PermissionCheckResponse.get_example_response(),
285 )
286 },
287 common_responses,
288 ),
289)
290async def get_user_permissions_api(
291 request: Request = None, token: dict = Depends(verify_token), db: AsyncSession = Depends(get_db)
292):
293 """Get all permissions for the current user"""
294 try:
295 user_id = token.get("sub")
296 result = await check_user_permissions(db, user_id, None)
298 return APIResponse(code=200, message="User permissions retrieved", data=result)
299 except HTTPException:
300 raise
301 except Exception:
302 raise HTTPException(status_code=500)