13 KiB
Exception Handling Guide
Overview
The LetzShop API uses a structured custom exception system to provide consistent, meaningful error responses across all endpoints. This guide covers how to work with exceptions, maintain the system, and extend it for new features.
Architecture
Exception Hierarchy
LetzShopException (Base)
├── ValidationException (400)
├── AuthenticationException (401)
├── AuthorizationException (403)
├── ResourceNotFoundException (404)
├── ConflictException (409)
├── BusinessLogicException (422)
├── RateLimitException (429)
└── ExternalServiceException (503)
File Structure
app/exceptions/
├── __init__.py # All exception exports
├── base.py # Base exception classes
├── handler.py # FastAPI exception handlers
├── auth.py # Authentication exceptions
├── admin.py # Admin operation exceptions
├── marketplace.py # Import/marketplace exceptions
├── product.py # Product management exceptions
├── shop.py # Shop management exceptions
└── stock.py # Stock management exceptions
Core Concepts
Exception Response Format
All custom exceptions return a consistent JSON structure:
{
"error_code": "PRODUCT_NOT_FOUND",
"message": "Product with ID 'ABC123' not found",
"status_code": 404,
"field": "product_id",
"details": {
"resource_type": "Product",
"identifier": "ABC123"
}
}
Properties
- error_code: Machine-readable identifier for programmatic handling
- message: Human-readable description for users
- status_code: HTTP status code
- field: Optional field name for validation errors
- details: Additional context data
Working with Exceptions
Basic Usage
from app.exceptions import ProductNotFoundException, ProductAlreadyExistsException
# Simple not found
raise ProductNotFoundException("ABC123")
# Conflict with existing resource
raise ProductAlreadyExistsException("ABC123")
Validation Exceptions
from app.exceptions import ValidationException, InvalidProductDataException
# Generic validation error
raise ValidationException("Invalid data format", field="price")
# Specific product validation
raise InvalidProductDataException(
"Price must be positive",
field="price",
details={"provided_price": -10}
)
Business Logic Exceptions
from app.exceptions import InsufficientStockException
# Detailed business logic error
raise InsufficientStockException(
gtin="1234567890",
location="warehouse_a",
requested=100,
available=50
)
Endpoint Implementation Patterns
Service Layer Integration
Map service layer ValueError exceptions to specific custom exceptions:
@router.post("/product")
def create_product(product: ProductCreate, db: Session = Depends(get_db)):
try:
result = product_service.create_product(db, product)
return result
except ValueError as e:
error_msg = str(e).lower()
if "already exists" in error_msg:
raise ProductAlreadyExistsException(product.product_id)
elif "invalid gtin" in error_msg:
raise InvalidProductDataException(str(e), field="gtin")
else:
raise ValidationException(str(e))
except Exception as e:
logger.error(f"Unexpected error: {str(e)}")
raise ValidationException("Failed to create product")
Error Message Parsing
Use regex patterns to extract context from service errors:
import re
try:
service.remove_stock(gtin, location, quantity)
except ValueError as e:
error_msg = str(e).lower()
if "insufficient stock" in error_msg:
# Extract quantities from error message
available_match = re.search(r'available[:\s]+(\d+)', error_msg)
requested_match = re.search(r'requested[:\s]+(\d+)', error_msg)
available = int(available_match.group(1)) if available_match else 0
requested = int(requested_match.group(1)) if requested_match else quantity
raise InsufficientStockException(gtin, location, requested, available)
Exception Re-raising
Always re-raise custom exceptions to maintain their specific type:
try:
shop = get_user_shop(shop_code, current_user, db)
# ... operations
except UnauthorizedShopAccessException:
raise # Re-raise without modification
except ValueError as e:
# Handle other ValueError cases
raise InvalidShopDataException(str(e))
Creating New Exceptions
Step 1: Define the Exception
Create new exceptions in the appropriate module:
# app/exceptions/payment.py
from .base import BusinessLogicException, ResourceNotFoundException
class PaymentNotFoundException(ResourceNotFoundException):
"""Raised when payment record is not found."""
def __init__(self, payment_id: str):
super().__init__(
resource_type="Payment",
identifier=payment_id,
message=f"Payment with ID '{payment_id}' not found",
error_code="PAYMENT_NOT_FOUND",
)
class InsufficientFundsException(BusinessLogicException):
"""Raised when account has insufficient funds."""
def __init__(self, required: float, available: float, account_id: str):
super().__init__(
message=f"Insufficient funds. Required: {required}, Available: {available}",
error_code="INSUFFICIENT_FUNDS",
details={
"required_amount": required,
"available_amount": available,
"account_id": account_id,
},
)
Step 2: Export in init.py
Add new exceptions to the exports:
# app/exceptions/__init__.py
from .payment import (
PaymentNotFoundException,
InsufficientFundsException,
)
__all__ = [
# ... existing exports
"PaymentNotFoundException",
"InsufficientFundsException",
]
Step 3: Use in Endpoints
Import and use the new exceptions:
from app.exceptions import PaymentNotFoundException, InsufficientFundsException
@router.post("/payments/{payment_id}/process")
def process_payment(payment_id: str, db: Session = Depends(get_db)):
try:
result = payment_service.process_payment(db, payment_id)
return result
except ValueError as e:
if "not found" in str(e).lower():
raise PaymentNotFoundException(payment_id)
elif "insufficient funds" in str(e).lower():
# Extract amounts from error message
raise InsufficientFundsException(100.0, 50.0, "account_123")
else:
raise ValidationException(str(e))
Testing Exceptions
Unit Tests
Test exception raising and properties:
import pytest
from app.exceptions import ProductNotFoundException, InsufficientStockException
def test_product_not_found_exception():
with pytest.raises(ProductNotFoundException) as exc_info:
raise ProductNotFoundException("ABC123")
exception = exc_info.value
assert exception.error_code == "PRODUCT_NOT_FOUND"
assert exception.status_code == 404
assert "ABC123" in exception.message
assert exception.details["identifier"] == "ABC123"
def test_insufficient_stock_exception():
exc = InsufficientStockException(
gtin="1234567890",
location="warehouse",
requested=100,
available=50
)
assert exc.error_code == "INSUFFICIENT_STOCK"
assert exc.status_code == 422
assert exc.details["requested_quantity"] == 100
assert exc.details["available_quantity"] == 50
Integration Tests
Test endpoint exception handling:
def test_create_duplicate_product(client, auth_headers):
# Create initial product
product_data = {"product_id": "TEST123", "name": "Test Product"}
client.post("/api/v1/product", json=product_data, headers=auth_headers)
# Attempt to create duplicate
response = client.post("/api/v1/product", json=product_data, headers=auth_headers)
assert response.status_code == 409
data = response.json()
assert data["error_code"] == "PRODUCT_ALREADY_EXISTS"
assert "TEST123" in data["message"]
assert data["details"]["product_id"] == "TEST123"
Best Practices
Do's
- Use specific exceptions rather than generic ValidationException
- Include relevant context in exception details
- Map service errors to appropriate custom exceptions
- Re-raise custom exceptions without modification
- Log errors before raising exceptions
- Extract data from error messages when possible
Don'ts
- Don't use HTTPException in endpoints (use custom exceptions)
- Don't lose exception context when mapping errors
- Don't include sensitive data in error messages
- Don't create exceptions for every possible error case
- Don't ignore the exception hierarchy when creating new exceptions
Error Message Guidelines
# Good: Specific, actionable message
raise InsufficientStockException(
gtin="1234567890",
location="warehouse_a",
requested=100,
available=50
)
# Bad: Generic, unhelpful message
raise ValidationException("Operation failed")
# Good: Include identifier context
raise ProductNotFoundException("ABC123")
# Bad: No context about what wasn't found
raise ValidationException("Not found")
Debugging and Logging
Exception Logging
The exception handler automatically logs all exceptions:
# Logs include structured data
logger.error(
f"Custom exception in {request.method} {request.url}: "
f"{exc.error_code} - {exc.message}",
extra={
"error_code": exc.error_code,
"status_code": exc.status_code,
"details": exc.details,
"url": str(request.url),
"method": request.method,
}
)
Adding Debug Information
Include additional context in exception details:
try:
result = complex_operation()
except ValueError as e:
raise BusinessLogicException(
message="Complex operation failed",
error_code="COMPLEX_OPERATION_FAILED",
details={
"original_error": str(e),
"operation_params": {"param1": value1, "param2": value2},
"timestamp": datetime.utcnow().isoformat(),
}
)
Frontend Integration
Error Code Handling
Frontend can handle errors programmatically:
// JavaScript example
try {
const response = await api.createProduct(productData);
} catch (error) {
switch (error.error_code) {
case 'PRODUCT_ALREADY_EXISTS':
showError('Product ID already exists. Please choose a different ID.');
break;
case 'INVALID_PRODUCT_DATA':
showFieldError(error.field, error.message);
break;
default:
showGenericError(error.message);
}
}
User-Friendly Messages
Map error codes to user-friendly messages:
const ERROR_MESSAGES = {
'PRODUCT_NOT_FOUND': 'The requested product could not be found.',
'INSUFFICIENT_STOCK': 'Not enough stock available for this operation.',
'UNAUTHORIZED_SHOP_ACCESS': 'You do not have permission to access this shop.',
'IMPORT_JOB_CANNOT_BE_CANCELLED': 'This import job cannot be cancelled at this time.',
};
Monitoring and Metrics
Error Tracking
Track exception frequency and patterns:
# Add to exception handler
from prometheus_client import Counter
exception_counter = Counter(
'api_exceptions_total',
'Total number of API exceptions',
['error_code', 'endpoint', 'method']
)
# In exception handler
exception_counter.labels(
error_code=exc.error_code,
endpoint=request.url.path,
method=request.method
).inc()
Health Monitoring
Monitor exception rates for system health:
- Track error rates per endpoint
- Alert on unusual exception patterns
- Monitor specific business logic errors
- Track client vs server error ratios
Migration Guide
From HTTPException to Custom Exceptions
Replace existing HTTPException usage:
# Before
from fastapi import HTTPException
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# After
from app.exceptions import ProductNotFoundException
if not product:
raise ProductNotFoundException(product_id)
Updating Service Layer
Enhance service layer error messages:
# Before
def get_product(db, product_id):
product = db.query(Product).filter(Product.id == product_id).first()
if not product:
raise ValueError("Not found")
return product
# After
def get_product(db, product_id):
product = db.query(Product).filter(Product.id == product_id).first()
if not product:
raise ValueError(f"Product not found: {product_id}")
return product
This exception system provides a robust foundation for error handling that scales with your application while maintaining consistency for both developers and frontend clients.