Implement user management endpoints in FastAPI #64
Conversation
|
CodeAnt AI is reviewing your PR. Thanks for using CodeAnt! 🎉We're free for open-source projects. if you're enjoying it, help us grow by sharing. Share on X · |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
Your free trial has ended. If you'd like to continue receiving code reviews, you can add a payment method here.
|
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. 📝 WalkthroughWalkthroughA new FastAPI router for user management is introduced with three endpoints: GET /profile retrieves current user details from Firebase Auth, PUT /profile updates user information and custom claims, and POST /bulk-action performs batch operations (disable/delete) on multiple users with admin-only access. Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Poem
Pre-merge checks and finishing touches✅ Passed checks (3 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Summary of ChangesHello @visz11, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request establishes a foundational user management system for the application by introducing a new module with several key API endpoints. It allows authenticated users to access and modify their personal profiles, while also providing administrators with powerful tools to manage user accounts in bulk. The implementation relies on Firebase Authentication to securely handle user data and ensures proper access control for sensitive operations. Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
Refacto PR SummaryImplemented comprehensive user management endpoints in FastAPI with Firebase Authentication integration to enable user profile management and administrative bulk operations. The implementation provides secure profile CRUD operations for authenticated users and admin-only bulk user management capabilities. Key Changes:
Change HighlightsClick to expand
Sequence DiagramsequenceDiagram
participant U as User
participant API as FastAPI
participant Auth as Firebase Auth
participant FB as Firebase Admin
U->>API: GET /users/profile
API->>Auth: Verify JWT token
Auth-->>API: User data
API->>FB: get_user(uid)
FB-->>API: User record + custom claims
API-->>U: Profile data
U->>API: PUT /users/profile
API->>Auth: Verify JWT token
API->>FB: update_user() + set_custom_user_claims()
FB-->>API: Success
API-->>U: Profile updated
Note over U,FB: Admin bulk operations
U->>API: POST /users/bulk-action
API->>Auth: Verify admin role
API->>FB: update_user(disabled=True) / delete_user()
FB-->>API: Operation results
API-->>U: Bulk operation results
Testing GuideClick to expand
|
|
Refacto is reviewing this PR. Please wait for the review comments to be posted. |
codeant-ai
bot
added
the
size:M
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces new user management endpoints. While the functionality is a good addition, there are several critical issues that need to be addressed. The most significant problems are KeyError bugs that will cause endpoints to fail due to incorrect dictionary key access ('id' instead of 'uid'). Additionally, exception handling is consistently too broad, which can mask bugs and leak sensitive information. I've also suggested improvements for authorization logic, data validation using Literal types, and safer access to user claims. Addressing these points will significantly improve the robustness and security of the new endpoints.
| async def get_user_profile(current_user: Dict[str, Any] = Depends(get_current_user)): | ||
| """Get detailed user profile information""" | ||
| try: | ||
| user_record = auth.get_user(current_user["id"]) |
There was a problem hiding this comment.
The current_user dictionary returned by the get_current_user dependency uses 'uid' as the key for the user ID, not 'id'. Accessing current_user["id"] will result in a KeyError and cause the endpoint to fail.
| user_record = auth.get_user(current_user["id"]) | |
| user_record = auth.get_user(current_user["uid"]) |
| ): | ||
| """Update current user's profile information""" | ||
| try: | ||
| user_record = auth.get_user(current_user["id"]) |
There was a problem hiding this comment.
Similar to the profile retrieval endpoint, the current_user dictionary uses 'uid' as the key for the user ID, not 'id'. This will cause a KeyError.
| user_record = auth.get_user(current_user["id"]) | |
| user_record = auth.get_user(current_user["uid"]) |
| "first_name": custom_claims["first_name"], | ||
| "last_name": custom_claims["last_name"], |
There was a problem hiding this comment.
Accessing custom claims directly with [] can cause a KeyError if a claim is not set for a user. It's safer to use the .get() method to avoid this and provide a graceful fallback (it will return null in the JSON if the key is missing).
| "first_name": custom_claims["first_name"], | |
| "last_name": custom_claims["last_name"], | |
| "first_name": custom_claims.get("first_name"), | |
| "last_name": custom_claims.get("last_name"), |
| except Exception as e: | ||
| raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) |
There was a problem hiding this comment.
Catching a generic Exception is too broad and can hide underlying issues. It's better to catch specific exceptions, like auth.UserNotFoundError. Additionally, exposing raw exception messages via str(e) can be a security risk. Provide a generic error message for unexpected server errors.
| except Exception as e: | |
| raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) | |
| except auth.UserNotFoundError: | |
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") | |
| except Exception: | |
| # It's good practice to log the actual error here for debugging purposes. | |
| raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="An internal error occurred.") |
| except Exception as e: | ||
| raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) |
There was a problem hiding this comment.
The exception handling is too broad. Catching specific exceptions like auth.UserNotFoundError would provide more accurate error responses. Also, returning HTTP_400_BAD_REQUEST for any server-side failure might be misleading; a 5xx error would be more appropriate for internal errors. Avoid exposing raw exception details.
| except Exception as e: | |
| raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) | |
| except auth.UserNotFoundError: | |
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") | |
| except Exception: | |
| # It's good practice to log the actual error here for debugging purposes. | |
| raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Failed to update profile.") |
| except Exception as e: | ||
| results.append({"user_id": user_id, "status": "error", "message": str(e)}) |
There was a problem hiding this comment.
The exception handling here is too broad. Catching specific exceptions like auth.UserNotFoundError will provide more meaningful error messages in the bulk operation results. Also, avoid exposing raw exception details in the response.
| except Exception as e: | |
| results.append({"user_id": user_id, "status": "error", "message": str(e)}) | |
| except auth.UserNotFoundError: | |
| results.append({"user_id": user_id, "status": "error", "message": "User not found"}) | |
| except Exception: | |
| results.append({"user_id": user_id, "status": "error", "message": "An unexpected error occurred."}) |
| @@ -0,0 +1,79 @@ | |||
| from fastapi import APIRouter, Depends, HTTPException, status | |||
| from .firebase_auth import firebase_auth | |||
There was a problem hiding this comment.
The firebase_auth import is unused in this file and can be removed to improve code cleanliness.
| from fastapi import APIRouter, Depends, HTTPException, status | ||
| from .firebase_auth import firebase_auth | ||
| from .dependencies import get_current_user, require_admin | ||
| from typing import Dict, Any, List |
There was a problem hiding this comment.
To enable strong typing for the action field in BulkUserRequest, Literal should be imported from the typing module.
| from typing import Dict, Any, List | |
| from typing import Dict, Any, List, Literal |
|
|
||
| class BulkUserRequest(BaseModel): | ||
| user_ids: List[str] | ||
| action: str |
There was a problem hiding this comment.
Using a plain string for the action field is prone to errors. By using Literal['disable', 'delete'], you make the API contract explicit and leverage FastAPI's built-in validation to ensure only valid actions are submitted.
| action: str | |
| action: Literal["disable", "delete"] |
| async def bulk_user_action( | ||
| request: BulkUserRequest, | ||
| current_user: Dict[str, Any] = Depends(get_current_user) | ||
| ): | ||
| """Perform bulk actions on multiple users (admin only)""" | ||
| if current_user.get("role") != "admin": | ||
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") |
There was a problem hiding this comment.
This endpoint performs a manual authorization check for the admin role. It's better to use the require_admin dependency, which is already imported. This approach centralizes authorization logic, makes the endpoint's intent clearer, and is more idiomatic for FastAPI.
| async def bulk_user_action( | |
| request: BulkUserRequest, | |
| current_user: Dict[str, Any] = Depends(get_current_user) | |
| ): | |
| """Perform bulk actions on multiple users (admin only)""" | |
| if current_user.get("role") != "admin": | |
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") | |
| async def bulk_user_action( | |
| request: BulkUserRequest, | |
| _: Dict[str, Any] = Depends(require_admin) | |
| ): | |
| """Perform bulk actions on multiple users (admin only)""" |
Nitpicks 🔍
|
Code Review: Authentication Service - Critical Reliability & Security IssuesPR Confidence Score: 🟥 1 / 5👍 Well Done
📁 Selected files for review (1)
🎯 Custom Instructions
📝 Additional Comments
🧰 Additional context used
|
![refacto-visz[bot]](https://avatars.githubusercontent.com/in/1889637?s=60&v=4){kind=small}
| "first_name": custom_claims["first_name"], | ||
| "last_name": custom_claims["last_name"], |
There was a problem hiding this comment.
Missing Custom Claims Null Safety
Direct dictionary access on custom_claims without null checking causes KeyError when claims are empty or missing. Service crashes with 500 error when users lack first_name/last_name claims, making authentication service unreliable for users without complete profile data.
"first_name": custom_claims.get("first_name", "") if custom_claims else "",
"last_name": custom_claims.get("last_name", "") if custom_claims else "",
Commitable Suggestion
| "first_name": custom_claims["first_name"], | |
| "last_name": custom_claims["last_name"], | |
| "first_name": custom_claims.get("first_name", "") if custom_claims else "", | |
| "last_name": custom_claims.get("last_name", "") if custom_claims else "", |
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- SRE-Error-Handling
Context References
get_user_profile function - no defensive checks for missing custom claimsinapp/auth/user_management.py
| @router.post("/bulk-action") | ||
| async def bulk_user_action( | ||
| request: BulkUserRequest, | ||
| current_user: Dict[str, Any] = Depends(get_current_user) | ||
| ): | ||
| """Perform bulk actions on multiple users (admin only)""" | ||
| if current_user.get("role") != "admin": | ||
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") |
There was a problem hiding this comment.
Authorization Race Condition
Authorization check occurs after function execution begins rather than using dependency injection. This creates a race condition where unauthorized users can trigger function execution before authorization validation, potentially allowing timing attacks or resource consumption before being blocked.
@router.post("/bulk-action")
async def bulk_user_action(
request: BulkUserRequest,
current_user: Dict[str, Any] = Depends(require_admin)
):
"""Perform bulk actions on multiple users (admin only)"""
Commitable Suggestion
| @router.post("/bulk-action") | |
| async def bulk_user_action( | |
| request: BulkUserRequest, | |
| current_user: Dict[str, Any] = Depends(get_current_user) | |
| ): | |
| """Perform bulk actions on multiple users (admin only)""" | |
| if current_user.get("role") != "admin": | |
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") | |
| @router.post("/bulk-action") | |
| async def bulk_user_action( | |
| request: BulkUserRequest, | |
| current_user: Dict[str, Any] = Depends(require_admin) | |
| ): | |
| """Perform bulk actions on multiple users (admin only)""" |
Standards
- CWE-862
- OWASP-A01
- NIST-SSDF-PW.1
Context References
require_admin dependency exists but not used - should replace inline checkinapp/auth/dependencies.py
![codeant-ai[bot]](https://avatars.githubusercontent.com/in/646884?s=60&v=4){kind=small}
| async def get_user_profile(current_user: Dict[str, Any] = Depends(get_current_user)): | ||
| """Get detailed user profile information""" | ||
| try: | ||
| user_record = auth.get_user(current_user["id"]) |
There was a problem hiding this comment.
Suggestion: The current user dict coming from the authentication dependency exposes a uid field (not id), so indexing current_user["id"] will raise a KeyError at runtime and prevent the profile from being fetched; use current_user["uid"] instead to look up the Firebase user. [logic error]
Severity Level: Minor
| user_record = auth.get_user(current_user["id"]) | |
| user_record = auth.get_user(current_user["uid"]) |
Why it matters? ⭐
The dependency get_current_user returns the dict produced by firebase_auth.verify_token, which contains "uid" (see app/auth/firebase_auth.py: lines returning {"uid": ...}) and does not provide an "id" key. Indexing current_user["id"] will raise a KeyError at runtime. Changing to current_user["uid"] fixes a real runtime bug.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** app/auth/user_management.py
**Line:** 26:26
**Comment:**
*Logic Error: The current user dict coming from the authentication dependency exposes a `uid` field (not `id`), so indexing `current_user["id"]` will raise a KeyError at runtime and prevent the profile from being fetched; use `current_user["uid"]` instead to look up the Firebase user.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.| custom_claims = auth.get_custom_user_claims(user_record.uid) | ||
| return { | ||
| "id": user_record.uid, | ||
| "email": user_record.email, | ||
| "first_name": custom_claims["first_name"], | ||
| "last_name": custom_claims["last_name"], |
There was a problem hiding this comment.
Suggestion: auth.get_custom_user_claims can return None or omit first_name/last_name for some users, and indexing like custom_claims["first_name"] will raise a KeyError; normalizing to an empty dict and using .get(..., "") avoids runtime failures for users without these claims. [possible bug]
Severity Level: Critical 🚨
| custom_claims = auth.get_custom_user_claims(user_record.uid) | |
| return { | |
| "id": user_record.uid, | |
| "email": user_record.email, | |
| "first_name": custom_claims["first_name"], | |
| "last_name": custom_claims["last_name"], | |
| custom_claims = auth.get_custom_user_claims(user_record.uid) or {} | |
| return { | |
| "id": user_record.uid, | |
| "email": user_record.email, | |
| "first_name": custom_claims.get("first_name", ""), | |
| "last_name": custom_claims.get("last_name", ""), |
Why it matters? ⭐
firebase_admin.auth.get_custom_user_claims can return None or omit keys. The current code indexes custom_claims["first_name"] / ["last_name"] directly, which will raise a KeyError (or TypeError if custom_claims is None). Normalizing to {} and using .get(..., "") prevents runtime errors for users without those claims. This is a defensive fix that prevents 500s for legitimate users.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** app/auth/user_management.py
**Line:** 27:32
**Comment:**
*Possible Bug: `auth.get_custom_user_claims` can return `None` or omit `first_name`/`last_name` for some users, and indexing like `custom_claims["first_name"]` will raise a KeyError; normalizing to an empty dict and using `.get(..., "")` avoids runtime failures for users without these claims.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.| ): | ||
| """Update current user's profile information""" | ||
| try: | ||
| user_record = auth.get_user(current_user["id"]) |
There was a problem hiding this comment.
Suggestion: As with the profile GET handler, the current user data produced by the auth dependency contains a uid field rather than id, so using current_user["id"] here will raise a KeyError and break profile updates; it should reference current_user["uid"] instead. [logic error]
Severity Level: Minor
| user_record = auth.get_user(current_user["id"]) | |
| user_record = auth.get_user(current_user["uid"]) |
Why it matters? ⭐
Same root cause as suggestion 1: get_current_user returns a dict with "uid" (from verify_token) not "id". The update_user_profile function uses current_user["id"] and will raise KeyError. Switching to current_user["uid"] resolves a real logic/runtime error.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** app/auth/user_management.py
**Line:** 47:47
**Comment:**
*Logic Error: As with the profile GET handler, the current user data produced by the auth dependency contains a `uid` field rather than `id`, so using `current_user["id"]` here will raise a KeyError and break profile updates; it should reference `current_user["uid"]` instead.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.| """Perform bulk actions on multiple users (admin only)""" | ||
| if current_user.get("role") != "admin": | ||
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") | ||
| results = [] |
There was a problem hiding this comment.
Suggestion: The bulk action endpoint silently accepts any action string and just returns an empty results list when the action is unsupported, which is misleading and hides misuse of the API; it should explicitly reject invalid actions with a 400 error before processing. [logic error]
Severity Level: Minor
| results = [] | |
| if request.action not in {"disable", "delete"}: | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Unsupported bulk action" | |
| ) |
Why it matters? ⭐
As written, unsupported actions are silently ignored and callers get an empty/partial results list which is misleading. Validating request.action early and returning HTTP 400 for unsupported actions is a correctness improvement and prevents clients from thinking the call succeeded when it did nothing.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** app/auth/user_management.py
**Line:** 68:68
**Comment:**
*Logic Error: The bulk action endpoint silently accepts any `action` string and just returns an empty `results` list when the action is unsupported, which is misleading and hides misuse of the API; it should explicitly reject invalid actions with a 400 error before processing.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
| for user_id in request.user_ids: | ||
| try: | ||
| if request.action == "disable": | ||
| auth.update_user(user_id, disabled=True) | ||
| results.append({"user_id": user_id, "status": "disabled"}) | ||
| elif request.action == "delete": | ||
| auth.delete_user(user_id) | ||
| results.append({"user_id": user_id, "status": "deleted"}) | ||
| except Exception as e: | ||
| results.append({"user_id": user_id, "status": "error", "message": str(e)}) |
There was a problem hiding this comment.
Sequential User Processing Bottleneck
Sequential Firebase auth operations in loop create O(n) blocking I/O pattern. Each user operation waits for Firebase API response before processing next user. Bulk operations with hundreds of users will cause significant response delays and poor scalability.
import asyncio
async def process_user(user_id: str, action: str):
try:
if action == "disable":
auth.update_user(user_id, disabled=True)
return {"user_id": user_id, "status": "disabled"}
elif action == "delete":
auth.delete_user(user_id)
return {"user_id": user_id, "status": "deleted"}
except Exception as e:
return {"user_id": user_id, "status": "error", "message": str(e)}
tasks = [process_user(user_id, request.action) for user_id in request.user_ids]
results = await asyncio.gather(*tasks, return_exceptions=True)
Commitable Suggestion
| for user_id in request.user_ids: | |
| try: | |
| if request.action == "disable": | |
| auth.update_user(user_id, disabled=True) | |
| results.append({"user_id": user_id, "status": "disabled"}) | |
| elif request.action == "delete": | |
| auth.delete_user(user_id) | |
| results.append({"user_id": user_id, "status": "deleted"}) | |
| except Exception as e: | |
| results.append({"user_id": user_id, "status": "error", "message": str(e)}) | |
| import asyncio | |
| async def process_user(user_id: str, action: str): | |
| try: | |
| if action == "disable": | |
| auth.update_user(user_id, disabled=True) | |
| return {"user_id": user_id, "status": "disabled"} | |
| elif action == "delete": | |
| auth.delete_user(user_id) | |
| return {"user_id": user_id, "status": "deleted"} | |
| except Exception as e: | |
| return {"user_id": user_id, "status": "error", "message": str(e)} | |
| tasks = [process_user(user_id, request.action) for user_id in request.user_ids] | |
| results = await asyncio.gather(*tasks, return_exceptions=True) |
Standards
- ISO-IEC-25010-Performance-Efficiency-Time-Behavior
- Optimization-Pattern-Async-Processing
- Algorithmic-Complexity-Linear-Optimization
| auth.set_custom_user_claims(user_record.uid, { | ||
| "first_name": update_data.first_name, | ||
| "last_name": update_data.last_name, | ||
| "role": current_user.get("role", "user") | ||
| }) |
There was a problem hiding this comment.
Custom Claims Data Loss
Complete custom claims replacement overwrites existing claims beyond first_name, last_name, and role. User loses additional custom claims causing feature degradation. Authorization and personalization data permanently lost during profile updates.
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- DbC-Postconditions
Context References
update_user_profile function - set_custom_user_claims replaces all claimsinapp/auth/user_management.py
| if request.action == "disable": | ||
| auth.update_user(user_id, disabled=True) | ||
| results.append({"user_id": user_id, "status": "disabled"}) | ||
| elif request.action == "delete": | ||
| auth.delete_user(user_id) | ||
| results.append({"user_id": user_id, "status": "deleted"}) |
There was a problem hiding this comment.
Missing Action Type Validation
Bulk action logic only handles 'disable' and 'delete' actions but lacks validation for invalid action types. Unrecognized actions silently skip processing without error indication, creating incomplete business rule enforcement.
Standards
- Business-Rule-Input-Validation
- Logic-Verification-Complete-Coverage
|
CodeAnt AI finished reviewing your PR. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (4)
app/auth/user_management.py (4)
3-3: Importedrequire_adminis not used.The
require_admindependency is imported but never utilized. The/bulk-actionendpoint manually checks the admin role at lines 66-67 instead of using this dependency.
11-14: Consider making fields optional for partial updates.All fields are currently required, forcing clients to provide
first_name, andlast_nameeven when updating only one field.🔎 Proposed refactor to support optional fields
class UserUpdateRequest(BaseModel): - email: EmailStr - first_name: str - last_name: str + email: EmailStr | None = None + first_name: str | None = None + last_name: str | None = NoneYou'll also need to update the
update_user_profilefunction to handle optional fields conditionally.
36-37: Improve exception handling specificity.Catching bare
Exceptionmasks all errors, including programming errors. Additionally, re-raising withoutfrom eloses the original traceback.🔎 Proposed improvement
- except Exception as e: - raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) + except auth.UserNotFoundError as e: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") from e + except Exception as e: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) from e
56-57: Improve exception handling specificity.Catching bare
Exceptionand re-raising withoutfrom eloses context. Consider handling specific Firebase exceptions.🔎 Proposed improvement
- except Exception as e: - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + except auth.UserNotFoundError as e: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") from e + except auth.EmailAlreadyExistsError as e: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already in use") from e + except Exception as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
app/auth/user_management.py
🧰 Additional context used
🧬 Code graph analysis (1)
app/auth/user_management.py (1)
app/auth/dependencies.py (1)
get_current_user(10-32)
🪛 Ruff (0.14.10)
app/auth/user_management.py
23-23: Do not perform function call Depends in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
36-36: Do not catch blind exception: Exception
(BLE001)
37-37: Within an except clause, raise exceptions with raise ... from err or raise ... from None to distinguish them from errors in exception handling
(B904)
43-43: Do not perform function call Depends in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
55-55: Consider moving this statement to an else block
(TRY300)
56-56: Do not catch blind exception: Exception
(BLE001)
57-57: Within an except clause, raise exceptions with raise ... from err or raise ... from None to distinguish them from errors in exception handling
(B904)
63-63: Do not perform function call Depends in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
77-77: Do not catch blind exception: Exception
(BLE001)
| class BulkUserRequest(BaseModel): | ||
| user_ids: List[str] | ||
| action: str |
There was a problem hiding this comment.
Validate the action field to prevent unintended operations.
The action field accepts any string value without validation, which is a security risk. Only "disable" and "delete" are handled in the code, but invalid actions would silently succeed without performing any operation.
🔎 Proposed fix using Literal for action validation
+from typing import Dict, Any, List, Literal
-from typing import Dict, Any, List
class BulkUserRequest(BaseModel):
user_ids: List[str]
- action: str
+ action: Literal["disable", "delete"]📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| class BulkUserRequest(BaseModel): | |
| user_ids: List[str] | |
| action: str | |
| from typing import Dict, Any, List, Literal | |
| class BulkUserRequest(BaseModel): | |
| user_ids: List[str] | |
| action: Literal["disable", "delete"] |
🤖 Prompt for AI Agents
In app/auth/user_management.py around lines 17 to 19, the BulkUserRequest.model
currently allows any string for action which can silently accept unsupported
operations; change the type of action to a constrained type (e.g.,
typing.Literal["disable","delete"] or a small Enum) so Pydantic will validate
input and reject invalid actions, update the import (from typing import Literal
or add Enum) and adjust any callers/tests if needed to use the new typed values;
ensure validation errors are surfaced to callers rather than allowing no-op
behavior.
| try: | ||
| user_record = auth.get_user(current_user["id"]) | ||
| custom_claims = auth.get_custom_user_claims(user_record.uid) | ||
| return { | ||
| "id": user_record.uid, | ||
| "email": user_record.email, | ||
| "first_name": custom_claims["first_name"], | ||
| "last_name": custom_claims["last_name"], | ||
| "is_active": not user_record.disabled, | ||
| "created_at": str(user_record.user_metadata.creation_timestamp) | ||
| } |
There was a problem hiding this comment.
Fix incorrect Firebase Admin SDK usage and handle missing custom claims.
Two critical issues:
-
Line 27:
auth.get_custom_user_claims()does not exist in the Firebase Admin SDK. Custom claims are accessed viauser_record.custom_claims. -
Lines 31-32: Direct dictionary access will raise
KeyErrorif custom claims areNoneor iffirst_name/last_namekeys don't exist.
🔎 Proposed fix
try:
user_record = auth.get_user(current_user["id"])
- custom_claims = auth.get_custom_user_claims(user_record.uid)
+ custom_claims = user_record.custom_claims or {}
return {
"id": user_record.uid,
"email": user_record.email,
- "first_name": custom_claims["first_name"],
- "last_name": custom_claims["last_name"],
+ "first_name": custom_claims.get("first_name"),
+ "last_name": custom_claims.get("last_name"),
"is_active": not user_record.disabled,
"created_at": str(user_record.user_metadata.creation_timestamp)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| try: | |
| user_record = auth.get_user(current_user["id"]) | |
| custom_claims = auth.get_custom_user_claims(user_record.uid) | |
| return { | |
| "id": user_record.uid, | |
| "email": user_record.email, | |
| "first_name": custom_claims["first_name"], | |
| "last_name": custom_claims["last_name"], | |
| "is_active": not user_record.disabled, | |
| "created_at": str(user_record.user_metadata.creation_timestamp) | |
| } | |
| try: | |
| user_record = auth.get_user(current_user["id"]) | |
| custom_claims = user_record.custom_claims or {} | |
| return { | |
| "id": user_record.uid, | |
| "email": user_record.email, | |
| "first_name": custom_claims.get("first_name"), | |
| "last_name": custom_claims.get("last_name"), | |
| "is_active": not user_record.disabled, | |
| "created_at": str(user_record.user_metadata.creation_timestamp) | |
| } |
🤖 Prompt for AI Agents
In app/auth/user_management.py around lines 25 to 35, replace the incorrect call
to auth.get_custom_user_claims(...) with reading user_record.custom_claims and
guard against None or missing keys; after fetching user_record =
auth.get_user(current_user["id"]), read claims = user_record.custom_claims or {}
and use claims.get("first_name") and claims.get("last_name") (or provide safe
defaults like None or empty string) when building the return dict, so the code
never raises KeyError if custom claims are absent.
| try: | ||
| user_record = auth.get_user(current_user["id"]) | ||
| if update_data.email != user_record.email: | ||
| auth.update_user(user_record.uid, email=update_data.email) | ||
| auth.set_custom_user_claims(user_record.uid, { | ||
| "first_name": update_data.first_name, | ||
| "last_name": update_data.last_name, | ||
| "role": current_user.get("role", "user") | ||
| }) |
There was a problem hiding this comment.
Risk of overwriting existing custom claims.
The set_custom_user_claims call replaces all custom claims with only first_name, last_name, and role. If other custom claims exist (e.g., permissions, metadata), they will be lost.
🔎 Proposed fix to preserve existing claims
try:
user_record = auth.get_user(current_user["id"])
if update_data.email != user_record.email:
auth.update_user(user_record.uid, email=update_data.email)
+
+ # Preserve existing custom claims
+ existing_claims = user_record.custom_claims or {}
+ updated_claims = {
+ **existing_claims,
+ "first_name": update_data.first_name,
+ "last_name": update_data.last_name,
+ "role": current_user.get("role", "user")
+ }
- auth.set_custom_user_claims(user_record.uid, {
- "first_name": update_data.first_name,
- "last_name": update_data.last_name,
- "role": current_user.get("role", "user")
- })
+ auth.set_custom_user_claims(user_record.uid, updated_claims)
return {"message": "Profile updated successfully"}🤖 Prompt for AI Agents
In app/auth/user_management.py around lines 46 to 54, the call to
set_custom_user_claims unconditionally overwrites all existing custom claims;
instead, fetch the current custom claims from user_record (or auth.get_user(...)
if not present), merge them with the new keys ("first_name", "last_name",
"role") so existing keys like "permissions" or "metadata" are preserved, and
then call set_custom_user_claims with the merged dictionary; ensure you handle
cases where user_record.custom_claims is None and validate/normalize types
before setting.
| @router.post("/bulk-action") | ||
| async def bulk_user_action( | ||
| request: BulkUserRequest, | ||
| current_user: Dict[str, Any] = Depends(get_current_user) | ||
| ): | ||
| """Perform bulk actions on multiple users (admin only)""" | ||
| if current_user.get("role") != "admin": | ||
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Use the require_admin dependency instead of manual role check.
The require_admin dependency is already imported but not used. Replace the manual role check with the dependency for consistency and better separation of concerns.
🔎 Proposed refactor
@router.post("/bulk-action")
async def bulk_user_action(
request: BulkUserRequest,
- current_user: Dict[str, Any] = Depends(get_current_user)
+ current_user: Dict[str, Any] = Depends(require_admin)
):
"""Perform bulk actions on multiple users (admin only)"""
- if current_user.get("role") != "admin":
- raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
results = []🧰 Tools
🪛 Ruff (0.14.10)
63-63: Do not perform function call Depends in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
🤖 Prompt for AI Agents
In app/auth/user_management.py around lines 60 to 67, replace the manual admin
role check with the existing require_admin dependency: remove the current_user
role-if-check and change the route signature to inject current_user via
Depends(require_admin) so the dependency enforces admin access; ensure the
function parameter uses the correct type and name (e.g., current_user: Dict[str,
Any] = Depends(require_admin)) and remove the now-unneeded HTTPException branch.
| results = [] | ||
| for user_id in request.user_ids: | ||
| try: | ||
| if request.action == "disable": | ||
| auth.update_user(user_id, disabled=True) | ||
| results.append({"user_id": user_id, "status": "disabled"}) | ||
| elif request.action == "delete": | ||
| auth.delete_user(user_id) | ||
| results.append({"user_id": user_id, "status": "deleted"}) | ||
| except Exception as e: | ||
| results.append({"user_id": user_id, "status": "error", "message": str(e)}) | ||
| return {"results": results} |
There was a problem hiding this comment.
Add safeguards for bulk operations.
Several concerns with the current implementation:
- No limit on bulk size: Processing unlimited users could cause timeouts or resource exhaustion.
- User deletion is permanent: No confirmation, soft-delete option, or audit trail for this destructive operation.
- Bare exception catching: Loses error context for debugging.
🔎 Proposed improvements
Add a limit check at the beginning:
# Add after the docstring
MAX_BULK_SIZE = 100
if len(request.user_ids) > MAX_BULK_SIZE:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Cannot process more than {MAX_BULK_SIZE} users at once"
)Consider more specific exception handling in the loop:
try:
if request.action == "disable":
auth.update_user(user_id, disabled=True)
results.append({"user_id": user_id, "status": "disabled"})
elif request.action == "delete":
auth.delete_user(user_id)
results.append({"user_id": user_id, "status": "deleted"})
+ except auth.UserNotFoundError as e:
+ results.append({"user_id": user_id, "status": "error", "message": "User not found"})
except Exception as e:
results.append({"user_id": user_id, "status": "error", "message": str(e)})For the deletion operation, consider implementing an audit log or requiring additional confirmation.
Committable suggestion skipped: line range outside the PR's diff.
🧰 Tools
🪛 Ruff (0.14.10)
77-77: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
In app/auth/user_management.py around lines 68 to 79, the bulk user operation
lacks safeguards: add a MAX_BULK_SIZE constant (e.g., 100) and check
len(request.user_ids) immediately to raise HTTPException 400 when exceeded;
replace the broad try/except with more specific exception handling (catch
auth-specific exceptions and fallback to Exception only to log full traceback)
and ensure errors are logged with context (user_id and exception) rather than
just returning str(e); change delete behavior to a safer approach by either
implementing soft-delete (marking a deleted flag), requiring an explicit
confirmation flag on the request for permanent deletes, and record audit entries
for deletes (user_id, actor, action, timestamp) before performing deletion;
ensure the response still collects per-user results but that destructive actions
enforce confirmation/audit and that failures are logged with full error context.
CodeAnt-AI Description
Add user profile endpoints and a bulk admin action endpoint
What Changed
Impact
✅ Users can update their email and name without support✅ Admins can disable or delete multiple accounts in one request✅ Requests surface account active status and creation time for clearer profiles💡 Usage Guide
Checking Your Pull Request
Every time you make a pull request, our system automatically looks through it. We check for security issues, mistakes in how you're setting up your infrastructure, and common code problems. We do this to make sure your changes are solid and won't cause any trouble later.
Talking to CodeAnt AI
Got a question or need a hand with something in your pull request? You can easily get in touch with CodeAnt AI right here. Just type the following in a comment on your pull request, and replace "Your question here" with whatever you want to ask:
This lets you have a chat with CodeAnt AI about your pull request, making it easier to understand and improve your code.
Example
Preserve Org Learnings with CodeAnt
You can record team preferences so CodeAnt AI applies them in future reviews. Reply directly to the specific CodeAnt AI suggestion (in the same thread) and replace "Your feedback here" with your input:
This helps CodeAnt AI learn and adapt to your team's coding style and standards.
Example
Retrigger review
Ask CodeAnt AI to review the PR again, by typing:
Check Your Repository Health
To analyze the health of your code repository, visit our dashboard at https://app.codeant.ai. This tool helps you identify potential issues and areas for improvement in your codebase, ensuring your repository maintains high standards of code health.
Summary by CodeRabbit
✏️ Tip: You can customize this high-level summary in your review settings.