API Exception Handling¶
Learn how to handle errors properly in your API endpoints using the boilerplate's built-in exceptions and patterns.
Quick Start¶
The boilerplate provides ready-to-use exceptions that return proper HTTP status codes:
from app.core.exceptions.http_exceptions import NotFoundException
@router.get("/{user_id}")
async def get_user(user_id: int, db: AsyncSession):
user = await crud_users.get(db=db, id=user_id)
if not user:
raise NotFoundException("User not found") # Returns 404
return user
That's it! The exception automatically becomes a proper JSON error response.
Built-in Exceptions¶
The boilerplate includes common HTTP exceptions you'll need:
NotFoundException (404)¶
from app.core.exceptions.http_exceptions import NotFoundException
@router.get("/{user_id}")
async def get_user(user_id: int):
user = await crud_users.get(db=db, id=user_id)
if not user:
raise NotFoundException("User not found")
return user
# Returns:
# Status: 404
# {"detail": "User not found"}
DuplicateValueException (409)¶
from app.core.exceptions.http_exceptions import DuplicateValueException
@router.post("/")
async def create_user(user_data: UserCreate):
if await crud_users.exists(db=db, email=user_data.email):
raise DuplicateValueException("Email already exists")
return await crud_users.create(db=db, object=user_data)
# Returns:
# Status: 409
# {"detail": "Email already exists"}
ForbiddenException (403)¶
from app.core.exceptions.http_exceptions import ForbiddenException
@router.delete("/{user_id}")
async def delete_user(
user_id: int,
current_user: Annotated[dict, Depends(get_current_user)]
):
if current_user["id"] != user_id and not current_user["is_superuser"]:
raise ForbiddenException("You can only delete your own account")
await crud_users.delete(db=db, id=user_id)
return {"message": "User deleted"}
# Returns:
# Status: 403
# {"detail": "You can only delete your own account"}
UnauthorizedException (401)¶
from app.core.exceptions.http_exceptions import UnauthorizedException
# This is typically used in the auth system, but you can use it too:
@router.get("/admin-only")
async def admin_endpoint():
# Some validation logic
if not user_is_admin:
raise UnauthorizedException("Admin access required")
return {"data": "secret admin data"}
# Returns:
# Status: 401
# {"detail": "Admin access required"}
Common Patterns¶
Check Before Create¶
@router.post("/", response_model=UserRead)
async def create_user(user_data: UserCreate, db: AsyncSession):
# Check email
if await crud_users.exists(db=db, email=user_data.email):
raise DuplicateValueException("Email already exists")
# Check username
if await crud_users.exists(db=db, username=user_data.username):
raise DuplicateValueException("Username already taken")
# Create user
return await crud_users.create(db=db, object=user_data)
# For public registration endpoints, consider rate limiting
# to prevent email enumeration attacks
Check Before Update¶
@router.patch("/{user_id}", response_model=UserRead)
async def update_user(
user_id: int,
user_data: UserUpdate,
db: AsyncSession
):
# Check if user exists
if not await crud_users.exists(db=db, id=user_id):
raise NotFoundException("User not found")
# Check for email conflicts (if email is being updated)
if user_data.email:
existing = await crud_users.get(db=db, email=user_data.email)
if existing and existing.id != user_id:
raise DuplicateValueException("Email already taken")
# Update user
return await crud_users.update(db=db, object=user_data, id=user_id)
Check Ownership¶
@router.get("/{post_id}")
async def get_post(
post_id: int,
current_user: Annotated[dict, Depends(get_current_user)],
db: AsyncSession
):
post = await crud_posts.get(db=db, id=post_id)
if not post:
raise NotFoundException("Post not found")
# Check if user owns the post or is admin
if post.author_id != current_user["id"] and not current_user["is_superuser"]:
raise ForbiddenException("You can only view your own posts")
return post
Validation Errors¶
FastAPI automatically handles Pydantic validation errors, but you can catch and customize them:
from fastapi import HTTPException
from pydantic import ValidationError
@router.post("/")
async def create_user(user_data: UserCreate):
try:
# If user_data fails validation, Pydantic raises ValidationError
# FastAPI automatically converts this to a 422 response
return await crud_users.create(db=db, object=user_data)
except ValidationError as e:
# You can catch and customize if needed
raise HTTPException(
status_code=400,
detail=f"Invalid data: {e.errors()}"
)
Standard HTTP Exceptions¶
For other status codes, use FastAPI's HTTPException:
from fastapi import HTTPException
# Bad Request (400)
@router.post("/")
async def create_something(data: dict):
if not data.get("required_field"):
raise HTTPException(
status_code=400,
detail="required_field is missing"
)
# Too Many Requests (429)
@router.post("/")
async def rate_limited_endpoint():
if rate_limit_exceeded():
raise HTTPException(
status_code=429,
detail="Rate limit exceeded. Try again later."
)
# Internal Server Error (500)
@router.get("/")
async def risky_endpoint():
try:
# Some operation that might fail
result = risky_operation()
return result
except Exception as e:
# Log the error
logger.error(f"Unexpected error: {e}")
raise HTTPException(
status_code=500,
detail="An unexpected error occurred"
)
Creating Custom Exceptions¶
If you need custom exceptions, follow the boilerplate's pattern:
# In app/core/exceptions/http_exceptions.py (add to existing file)
from fastapi import HTTPException
class PaymentRequiredException(HTTPException):
"""402 Payment Required"""
def __init__(self, detail: str = "Payment required"):
super().__init__(status_code=402, detail=detail)
class TooManyRequestsException(HTTPException):
"""429 Too Many Requests"""
def __init__(self, detail: str = "Too many requests"):
super().__init__(status_code=429, detail=detail)
# Use them in your endpoints
from app.core.exceptions.http_exceptions import PaymentRequiredException
@router.get("/premium-feature")
async def premium_feature(current_user: dict):
if current_user["tier"] == "free":
raise PaymentRequiredException("Upgrade to access this feature")
return {"data": "premium content"}
Error Response Format¶
All exceptions return consistent JSON responses:
For validation errors (422), you get more detail:
{
"detail": [
{
"type": "missing",
"loc": ["body", "email"],
"msg": "Field required",
"input": null
}
]
}
Global Exception Handling¶
The boilerplate includes global exception handlers. You can add your own in main.py:
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
app = FastAPI()
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
"""Handle ValueError exceptions globally"""
return JSONResponse(
status_code=400,
content={"detail": f"Invalid value: {str(exc)}"}
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
"""Catch-all exception handler"""
# Log the error
logger.error(f"Unhandled exception: {exc}")
return JSONResponse(
status_code=500,
content={"detail": "An unexpected error occurred"}
)
Security Considerations¶
Authentication Endpoints - Use Generic Messages¶
For security, authentication endpoints should use generic error messages to prevent information disclosure:
# SECURITY: Don't reveal if username exists
@router.post("/login")
async def login(credentials: LoginCredentials):
user = await crud_users.get(db=db, username=credentials.username)
# Don't do this - reveals if username exists
# if not user:
# raise NotFoundException("User not found")
# if not verify_password(credentials.password, user.hashed_password):
# raise UnauthorizedException("Invalid password")
# Do this - generic message for all auth failures
if not user or not verify_password(credentials.password, user.hashed_password):
raise UnauthorizedException("Invalid username or password")
return create_access_token(user.id)
# SECURITY: Don't reveal if email is registered during password reset
@router.post("/forgot-password")
async def forgot_password(email: str):
user = await crud_users.get(db=db, email=email)
# Don't do this - reveals if email exists
# if not user:
# raise NotFoundException("Email not found")
# Do this - always return success message
if user:
await send_password_reset_email(user.email)
# Always return the same message
return {"message": "If the email exists, a reset link has been sent"}
Resource Access - Be Specific When Safe¶
For non-auth operations, specific messages help developers:
# Safe to be specific for resource operations
@router.get("/{post_id}")
async def get_post(
post_id: int,
current_user: Annotated[dict, Depends(get_current_user)]
):
post = await crud_posts.get(db=db, id=post_id)
if not post:
raise NotFoundException("Post not found") # Safe to be specific
if post.author_id != current_user["id"]:
# Don't reveal post exists if user can't access it
raise NotFoundException("Post not found") # Generic, not "Access denied"
return post
Best Practices¶
1. Use Specific Exceptions (When Safe)¶
# Good for non-sensitive operations
if not user:
raise NotFoundException("User not found")
# Good for validation errors
raise DuplicateValueException("Username already taken")
2. Use Generic Messages for Security¶
# Good for authentication
raise UnauthorizedException("Invalid username or password")
# Good for authorization (don't reveal resource exists)
raise NotFoundException("Resource not found") # Instead of "Access denied"
3. Check Permissions Early¶
@router.delete("/{user_id}")
async def delete_user(
user_id: int,
current_user: Annotated[dict, Depends(get_current_user)]
):
# Check permission first
if current_user["id"] != user_id:
raise ForbiddenException("Cannot delete other users")
# Then check if user exists
if not await crud_users.exists(db=db, id=user_id):
raise NotFoundException("User not found")
await crud_users.delete(db=db, id=user_id)
4. Log Important Errors¶
import logging
logger = logging.getLogger(__name__)
@router.post("/")
async def create_user(user_data: UserCreate):
try:
return await crud_users.create(db=db, object=user_data)
except Exception as e:
logger.error(f"Failed to create user: {e}")
raise HTTPException(status_code=500, detail="User creation failed")
Testing Exceptions¶
Test that your endpoints raise the right exceptions:
import pytest
from httpx import AsyncClient
@pytest.mark.asyncio
async def test_user_not_found(client: AsyncClient):
response = await client.get("/api/v1/users/99999")
assert response.status_code == 404
assert "User not found" in response.json()["detail"]
@pytest.mark.asyncio
async def test_duplicate_email(client: AsyncClient):
# Create a user
await client.post("/api/v1/users/", json={
"name": "Test User",
"username": "test1",
"email": "test@example.com",
"password": "Password123!"
})
# Try to create another with same email
response = await client.post("/api/v1/users/", json={
"name": "Test User 2",
"username": "test2",
"email": "test@example.com", # Same email
"password": "Password123!"
})
assert response.status_code == 409
assert "Email already exists" in response.json()["detail"]
What's Next¶
Now that you understand error handling: - Versioning - Learn how to version your APIs - Database CRUD - Understand the database operations - Authentication - Add user authentication to your APIs
Proper error handling makes your API much more user-friendly and easier to debug!