Error Handling and Validation

Decision Tree for Status Codes:

Did the request succeed?
├─ Yes → Was a resource created?
│  ├─ Yes → 201 Created (include Location header)
│  └─ No → Is there a response body?
│     ├─ Yes → 200 OK
│     └─ No → 204 No Content
│
└─ No → Is the problem with the client or server?
   ├─ Client problem (4xx)
   │  ├─ Authentication missing/invalid? → 401 Unauthorized
   │  ├─ Authenticated but lacks permission? → 403 Forbidden
   │  ├─ Resource not found? → 404 Not Found
   │  ├─ Invalid JSON or malformed request? → 400 Bad Request
   │  ├─ Validation failed (email invalid, etc.)? → 422 Unprocessable Entity
   │  ├─ Conflict (duplicate email, version mismatch)? → 409 Conflict
   │  ├─ Rate limit exceeded? → 429 Too Many Requests
   │  └─ Other client error → 400 Bad Request
   │
   └─ Server problem (5xx)
      ├─ Unexpected exception? → 500 Internal Server Error
      ├─ Database/external service down? → 503 Service Unavailable
      ├─ External API timeout? → 504 Gateway Timeout
      └─ Other server error → 500 Internal Server Error

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, field_validator

app = FastAPI()


class UserCreate(BaseModel):
    username: str
    email: str
    age: int

    @field_validator('email')
    @classmethod
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('Invalid email format')
        return v


@app.post("/api/users", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    """
    Returns 201 Created when resource is successfully created
    Location header points to new resource
    """
    # Check for duplicate (409 Conflict)
    existing = db.query(User).filter(User.email == user.email).first()
    if existing:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="User with this email already exists"
        )

    # Create user
    new_user = User(**user.model_dump())
    db.add(new_user)
    db.commit()

    return {
        "id": new_user.id,
        "username": new_user.username,
        "email": new_user.email
    }
    # Note: Add Location header manually via Response.headers["Location"]


@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    """Returns 200 OK with user data, or 404 Not Found"""
    user = db.query(User).filter(User.id == user_id).first()

    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )

    return user  # 200 OK


@app.delete("/api/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
    """Returns 204 No Content (no response body for DELETE)"""
    user = db.query(User).filter(User.id == user_id).first()

    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )

    db.delete(user)
    db.commit()
    # No return statement = 204 No Content


# Validation error → 422 Unprocessable Entity (automatic with Pydantic)
# POST /api/users with {"username": "john", "email": "not-an-email", "age": 25}
# Response: 422 with {"detail": [{"loc": ["body", "email"], "msg": "Invalid email format"}]}

from pydantic import BaseModel, ValidationError, field_validator

class UserCreate(BaseModel):
    username: str
    email: str
    age: int

    @field_validator('username')
    @classmethod
    def validate_username(cls, v):
        if len(v) < 3:
            raise ValueError('Username must be at least 3 characters')
        return v

    @field_validator('email')
    @classmethod
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('Invalid email format')
        return v

    @field_validator('age')
    @classmethod
    def validate_age(cls, v):
        if v < 13:
            raise ValueError('Must be at least 13 years old')
        return v


@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
    """Handle Pydantic validation errors with field-level details"""

    # Convert Pydantic errors to field-level format
    errors = {}
    for error in exc.errors():
        field = error['loc'][-1]  # Get field name
        errors[field] = {
            "message": error['msg'],
            "type": error['type']
        }

    problem = create_problem_detail(
        type_suffix="validation-failed",
        title="Validation Failed",
        status_code=422,
        detail="One or more fields failed validation",
        instance=request.url.path,
        errors=errors  # Custom field with field-level errors
    )

    return JSONResponse(
        status_code=422,
        content=problem.model_dump(exclude_none=True),
        headers={"Content-Type": "application/problem+json"}
    )


# POST /api/users
# Body: {"username": "ab", "email": "invalid", "age": 10}
# Response:
{
    "type": "https://api.example.com/errors/validation-failed",
    "title": "Validation Failed",
    "status": 422,
    "detail": "One or more fields failed validation",
    "instance": "/api/users",
    "errors": {
        "username": {
            "message": "Value error, Username must be at least 3 characters",
            "type": "value_error"
        },
        "email": {
            "message": "Value error, Invalid email format",
            "type": "value_error"
        },
        "age": {
            "message": "Value error, Must be at least 13 years old",
            "type": "value_error"
        }
    }
}

from fastapi import FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, field_validator
from typing import List, Dict, Any

app = FastAPI()


class ValidationErrorDetail(BaseModel):
    """Structured validation error for a single field"""
    field: str  # Field name (e.g., "email")
    message: str  # Human-readable error message
    code: str  # Machine-readable error code
    value: Any = None  # The invalid value (optional, omit for sensitive fields)


class ValidationErrorResponse(BaseModel):
    """Complete validation error response"""
    type: str = "https://api.example.com/errors/validation-failed"
    title: str = "Validation Failed"
    status: int = 422
    detail: str
    instance: str
    errors: List[ValidationErrorDetail]


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
    request: Request,
    exc: RequestValidationError
):
    """Convert Pydantic validation errors to structured format"""

    errors = []
    for error in exc.errors():
        # Get field path (e.g., ["body", "email"] → "email")
        field_path = ".".join(str(loc) for loc in error['loc'][1:])

        # Map Pydantic error types to custom error codes
        error_code_map = {
            "missing": "FIELD_REQUIRED",
            "value_error": "INVALID_VALUE",
            "string_too_short": "INVALID_VALUE",
            "string_too_long": "INVALID_VALUE",
            "string_pattern_mismatch": "INVALID_VALUE",
            "greater_than_equal": "INVALID_VALUE",
            "less_than_equal": "INVALID_VALUE",
            "int_parsing": "INVALID_TYPE",
        }

        error_code = error_code_map.get(
            error['type'],
            "VALIDATION_ERROR"
        )

        errors.append(ValidationErrorDetail(
            field=field_path,
            message=error['msg'],
            code=error_code,
            value=error.get('ctx', {}).get('given')  # Include invalid value
        ))

    response = ValidationErrorResponse(
        detail=f"{len(errors)} validation error(s) occurred",
        instance=request.url.path,
        errors=errors
    )

    return JSONResponse(
        status_code=422,
        content=response.model_dump(exclude_none=True),
        headers={"Content-Type": "application/problem+json"}
    )


class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=20)
    email: str = Field(..., pattern=r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
    age: int = Field(..., ge=13, le=120)
    website: str = Field(None, pattern=r'^https?://')

    @field_validator('username')
    @classmethod
    def username_alphanumeric(cls, v):
        if not v.replace('_', '').replace('-', '').isalnum():
            raise ValueError('Username must contain only letters, numbers, hyphens, and underscores')
        return v


@app.post("/api/users")
async def create_user(user: UserCreate):
    return {"message": "User created", "user": user.model_dump()}

from pydantic import BaseModel, Field, field_validator

# Define custom error messages with Field
class UserCreate(BaseModel):
    username: str = Field(
        ...,
        min_length=3,
        max_length=20,
        description="Username between 3-20 characters"
    )
    email: str = Field(
        ...,
        description="Valid email address"
    )
    password: str = Field(
        ...,
        min_length=8,
        description="Password at least 8 characters"
    )

    @field_validator('username')
    @classmethod
    def username_valid(cls, v):
        if not v[0].isalpha():
            raise ValueError('Username must start with a letter')
        if '__' in v:
            raise ValueError('Username cannot contain consecutive underscores')
        return v

    @field_validator('password')
    @classmethod
    def password_strong(cls, v):
        if not any(c.isupper() for c in v):
            raise ValueError('Password must contain at least one uppercase letter')
        if not any(c.isdigit() for c in v):
            raise ValueError('Password must contain at least one number')
        if not any(c in '!@#$%^&*()' for c in v):
            raise ValueError('Password must contain at least one special character (!@#$%^&*())')
        return v


# Result: Clear, specific error messages for each validation rule
# "Username must start with a letter"
# "Password must contain at least one uppercase letter"
# "Password must contain at least one number"

from enum import Enum
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI()


class ErrorCode(str, Enum):
    """Centralized error code catalog"""

    # Authentication errors (1xxx)
    AUTH_TOKEN_MISSING = "AUTH_TOKEN_MISSING"
    AUTH_TOKEN_INVALID = "AUTH_TOKEN_INVALID"
    AUTH_TOKEN_EXPIRED = "AUTH_TOKEN_EXPIRED"
    AUTH_CREDENTIALS_INVALID = "AUTH_CREDENTIALS_INVALID"

    # Authorization errors (2xxx)
    AUTHZ_INSUFFICIENT_PERMISSIONS = "AUTHZ_INSUFFICIENT_PERMISSIONS"
    AUTHZ_RESOURCE_NOT_OWNED = "AUTHZ_RESOURCE_NOT_OWNED"

    # User errors (3xxx)
    USER_NOT_FOUND = "USER_NOT_FOUND"
    USER_ALREADY_EXISTS = "USER_ALREADY_EXISTS"
    USER_EMAIL_DUPLICATE = "USER_EMAIL_DUPLICATE"
    USER_ACCOUNT_SUSPENDED = "USER_ACCOUNT_SUSPENDED"

    # Validation errors (4xxx)
    VALIDATION_FAILED = "VALIDATION_FAILED"
    VALIDATION_EMAIL_INVALID = "VALIDATION_EMAIL_INVALID"
    VALIDATION_AGE_TOO_YOUNG = "VALIDATION_AGE_TOO_YOUNG"

    # Payment errors (5xxx)
    PAYMENT_FAILED = "PAYMENT_FAILED"
    PAYMENT_INSUFFICIENT_FUNDS = "PAYMENT_INSUFFICIENT_FUNDS"
    PAYMENT_CARD_DECLINED = "PAYMENT_CARD_DECLINED"
    PAYMENT_CARD_EXPIRED = "PAYMENT_CARD_EXPIRED"

    # Rate limiting (6xxx)
    RATE_LIMIT_EXCEEDED = "RATE_LIMIT_EXCEEDED"

    # Resource conflicts (7xxx)
    RESOURCE_CONFLICT = "RESOURCE_CONFLICT"
    RESOURCE_LOCKED = "RESOURCE_LOCKED"
    RESOURCE_VERSION_MISMATCH = "RESOURCE_VERSION_MISMATCH"


class APIError(BaseModel):
    """Standardized error response with error code"""
    error_code: ErrorCode
    message: str  # Human-readable message
    status_code: int
    details: dict = None  # Optional additional context


class APIException(HTTPException):
    """Custom exception with error code"""
    def __init__(
        self,
        error_code: ErrorCode,
        message: str,
        status_code: int = status.HTTP_400_BAD_REQUEST,
        details: dict = None
    ):
        self.error_code = error_code
        self.message = message
        self.details = details
        super().__init__(status_code=status_code, detail=message)


@app.exception_handler(APIException)
async def api_exception_handler(request, exc: APIException):
    """Handler for custom API exceptions"""
    error_response = APIError(
        error_code=exc.error_code,
        message=exc.message,
        status_code=exc.status_code,
        details=exc.details
    )
    return JSONResponse(
        status_code=exc.status_code,
        content=error_response.model_dump(exclude_none=True)
    )


@app.post("/api/users")
async def create_user(email: str, username: str):
    """Create user with specific error codes"""

    # Check for duplicate email
    if user_exists_by_email(email):
        raise APIException(
            error_code=ErrorCode.USER_EMAIL_DUPLICATE,
            message=f"A user with email '{email}' already exists",
            status_code=status.HTTP_409_CONFLICT,
            details={"email": email}
        )

    # Check for duplicate username
    if user_exists_by_username(username):
        raise APIException(
            error_code=ErrorCode.USER_ALREADY_EXISTS,
            message=f"Username '{username}' is already taken",
            status_code=status.HTTP_409_CONFLICT,
            details={"username": username}
        )

    return {"message": "User created"}


@app.post("/api/payments")
async def process_payment(amount: int, card_token: str):
    """Process payment with detailed error codes"""

    result = charge_card(card_token, amount)

    if not result.success:
        # Map payment gateway errors to our error codes
        error_map = {
            "insufficient_funds": ErrorCode.PAYMENT_INSUFFICIENT_FUNDS,
            "card_declined": ErrorCode.PAYMENT_CARD_DECLINED,
            "expired_card": ErrorCode.PAYMENT_CARD_EXPIRED,
        }

        error_code = error_map.get(
            result.failure_reason,
            ErrorCode.PAYMENT_FAILED
        )

        raise APIException(
            error_code=error_code,
            message=f"Payment failed: {result.failure_reason}",
            status_code=status.HTTP_402_PAYMENT_REQUIRED,
            details={
                "amount": amount,
                "reason": result.failure_reason
            }
        )

    return {"status": "success", "transaction_id": result.transaction_id}

// Frontend TypeScript example
async function createUser(email: string, username: string) {
    try {
        const response = await fetch('/api/users', {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({ email, username })
        });

        if (!response.ok) {
            const error = await response.json();

            // Handle specific error codes
            switch (error.error_code) {
                case 'USER_EMAIL_DUPLICATE':
                    showError('This email is already registered. Please log in or use a different email.');
                    highlightField('email');
                    break;

                case 'USER_ALREADY_EXISTS':
                    showError('This username is taken. Please choose another.');
                    highlightField('username');
                    break;

                case 'VALIDATION_FAILED':
                    showValidationErrors(error.details);
                    break;

                default:
                    showError('An unexpected error occurred. Please try again.');
            }

            return;
        }

        showSuccess('Account created successfully!');
    } catch (err) {
        showError('Network error. Please check your connection.');
    }
}


// Payment handling example
async function processPayment(amount: number, cardToken: string) {
    const response = await fetch('/api/payments', {
        method: 'POST',
        body: JSON.stringify({ amount, card_token: cardToken })
    });

    if (!response.ok) {
        const error = await response.json();

        switch (error.error_code) {
            case 'PAYMENT_INSUFFICIENT_FUNDS':
                showError('Insufficient funds. Please add money to your account.');
                showButton('Add Funds');
                break;

            case 'PAYMENT_CARD_DECLINED':
                showError('Your card was declined. Please contact your bank.');
                break;

            case 'PAYMENT_CARD_EXPIRED':
                showError('Your card has expired. Please update your payment method.');
                showButton('Update Card');
                break;

            default:
                showError('Payment failed. Please try again or use a different card.');
        }
    }
}

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
import logging
import traceback
import os
import uuid

app = FastAPI()
logger = logging.getLogger(__name__)

# Detect environment
IS_PRODUCTION = os.getenv("ENV") == "production"


@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Global exception handler with environment-aware detail level"""

    # Generate unique request ID for tracking
    request_id = str(uuid.uuid4())

    # Log full error details internally (always, regardless of environment)
    logger.error(
        f"Request ID: {request_id} | "
        f"Path: {request.url.path} | "
        f"Method: {request.method} | "
        f"Error: {str(exc)} | "
        f"Traceback: {traceback.format_exc()}"
    )

    if IS_PRODUCTION:
        # Production: Sanitized error response
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={
                "error_code": "INTERNAL_ERROR",
                "message": "An unexpected error occurred. Please try again later.",
                "request_id": request_id,  # Support can use this to look up logs
                "support_url": "https://support.example.com/contact"
            }
        )
    else:
        # Development: Verbose error response with debugging info
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={
                "error_code": "INTERNAL_ERROR",
                "message": str(exc),
                "exception_type": type(exc).__name__,
                "request_id": request_id,
                "traceback": traceback.format_exc().split('\n'),
                "request": {
                    "path": request.url.path,
                    "method": request.method,
                    "headers": dict(request.headers),
                    "query_params": dict(request.query_params)
                }
            }
        )


# Development error example response:
{
    "error_code": "INTERNAL_ERROR",
    "message": "division by zero",
    "exception_type": "ZeroDivisionError",
    "request_id": "req_abc123",
    "traceback": [
        "Traceback (most recent call last):",
        "  File 'main.py', line 120, in process_payment",
        "    result = amount / 0",
        "ZeroDivisionError: division by zero"
    ],
    "request": {
        "path": "/api/payments",
        "method": "POST",
        "headers": {...},
        "query_params": {}
    }
}


# Production error example response:
{
    "error_code": "INTERNAL_ERROR",
    "message": "An unexpected error occurred. Please try again later.",
    "request_id": "req_abc123",
    "support_url": "https://support.example.com/contact"
}

from fastapi import FastAPI, Request, Response, HTTPException, status
from fastapi.responses import JSONResponse
import time
import redis

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Rate limit: 100 requests per minute per user
RATE_LIMIT = 100
RATE_WINDOW = 60  # seconds


def check_rate_limit(user_id: str) -> tuple[bool, int, int]:
    """
    Check if user has exceeded rate limit.
    Returns: (allowed, remaining, reset_timestamp)
    """
    key = f"rate_limit:{user_id}"
    now = int(time.time())
    window_start = now - RATE_WINDOW

    # Remove old requests outside the window
    redis_client.zremrangebyscore(key, 0, window_start)

    # Count requests in current window
    request_count = redis_client.zcard(key)

    if request_count >= RATE_LIMIT:
        # Rate limit exceeded
        oldest_request = redis_client.zrange(key, 0, 0, withscores=True)
        if oldest_request:
            reset_time = int(oldest_request[0][1]) + RATE_WINDOW
        else:
            reset_time = now + RATE_WINDOW

        return False, 0, reset_time

    # Add current request
    redis_client.zadd(key, {str(now): now})
    redis_client.expire(key, RATE_WINDOW)

    remaining = RATE_LIMIT - request_count - 1
    reset_time = now + RATE_WINDOW

    return True, remaining, reset_time


@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    """Apply rate limiting to all requests"""

    # Get user ID from token (simplified)
    user_id = request.headers.get("X-User-ID", "anonymous")

    # Check rate limit
    allowed, remaining, reset_time = check_rate_limit(user_id)

    if not allowed:
        # Rate limit exceeded
        retry_after = reset_time - int(time.time())

        return JSONResponse(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            content={
                "error_code": "RATE_LIMIT_EXCEEDED",
                "message": f"Rate limit exceeded. Please retry after {retry_after} seconds.",
                "retry_after": retry_after
            },
            headers={
                "X-RateLimit-Limit": str(RATE_LIMIT),
                "X-RateLimit-Remaining": "0",
                "X-RateLimit-Reset": str(reset_time),
                "Retry-After": str(retry_after)  # Standard HTTP header
            }
        )

    # Process request
    response = await call_next(request)

    # Add rate limit headers to successful responses
    response.headers["X-RateLimit-Limit"] = str(RATE_LIMIT)
    response.headers["X-RateLimit-Remaining"] = str(remaining)
    response.headers["X-RateLimit-Reset"] = str(reset_time)

    return response

# Install: pip install tenacity requests

import requests
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log,
    after_log
)
import logging

# Configure logging
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)


# Basic retry with exponential backoff
@retry(
    stop=stop_after_attempt(3),  # Max 3 attempts
    wait=wait_exponential(multiplier=1, min=1, max=10),  # 1s, 2s, 4s, 8s, 10s (max)
    before_sleep=before_sleep_log(logger, logging.WARNING),
    after=after_log(logger, logging.INFO)
)
def fetch_user(user_id: int):
    """
    Fetch user with automatic retries.

    Retry behavior:
    - Attempt 1: Immediate
    - Attempt 2: Wait 1 second
    - Attempt 3: Wait 2 seconds
    - If all fail: Raise exception
    """
    response = requests.get(
        f"https://api.example.com/users/{user_id}",
        headers={"Authorization": "Bearer token"},
        timeout=10
    )
    response.raise_for_status()  # Raise exception for 4xx/5xx
    return response.json()


# Usage
try:
    user = fetch_user(123)
    print(f"Success: {user}")
except Exception as e:
    print(f"Failed after retries: {e}")

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception
from requests.exceptions import RequestException, Timeout, ConnectionError
import requests


class RateLimitError(Exception):
    """Custom exception for rate limit errors"""
    def __init__(self, retry_after: int):
        self.retry_after = retry_after
        super().__init__(f"Rate limited. Retry after {retry_after} seconds")


def should_retry(exception):
    """
    Determine if exception is retryable.

    Retry:
    - Network errors (timeout, connection refused)
    - 5xx server errors
    - 429 rate limit

    Don't retry:
    - 4xx client errors (bad request, unauthorized, not found)
    """
    if isinstance(exception, (Timeout, ConnectionError)):
        return True  # Network errors - always retry

    if isinstance(exception, RateLimitError):
        return True  # Rate limit - retry with backoff

    if isinstance(exception, requests.HTTPError):
        if 500 <= exception.response.status_code < 600:
            return True  # Server error - retry
        else:
            return False  # Client error - don't retry

    return False


@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception(should_retry),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
def api_call_with_smart_retry(endpoint: str):
    """
    API call that only retries transient failures.

    Returns immediately on:
    - 400 Bad Request (don't retry, fix the request)
    - 401 Unauthorized (don't retry, get valid token)
    - 404 Not Found (don't retry, resource doesn't exist)

    Retries on:
    - Network timeout/errors
    - 429 Too Many Requests
    - 500 Internal Server Error
    - 503 Service Unavailable
    """
    response = requests.get(
        f"https://api.example.com/{endpoint}",
        headers={"Authorization": "Bearer token"},
        timeout=10
    )

    # Check for rate limiting
    if response.status_code == 429:
        retry_after = int(response.headers.get("Retry-After", 60))
        raise RateLimitError(retry_after)

    # Raise exception for any error status
    response.raise_for_status()

    return response.json()


# Example: Calling the API
try:
    data = api_call_with_smart_retry("products")
    print(f"Success: {len(data)} products")

except requests.HTTPError as e:
    if e.response.status_code == 404:
        print("Resource not found (no retries attempted)")
    elif e.response.status_code == 401:
        print("Unauthorized (no retries attempted)")
    else:
        print(f"HTTP error after retries: {e}")

except RateLimitError as e:
    print(f"Rate limited after all retries: {e}")

except Exception as e:
    print(f"Request failed after retries: {e}")

from datetime import datetime, timezone
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception
import requests
import time


def wait_from_retry_after_header(retry_state):
    """
    Custom wait function that respects Retry-After header.
    Falls back to exponential backoff if header not present.
    """
    exception = retry_state.outcome.exception()

    if isinstance(exception, RateLimitError) and exception.retry_after:
        # Use server's Retry-After value
        wait_seconds = exception.retry_after
        logger.info(f"Rate limited. Waiting {wait_seconds}s as specified by server")
        return wait_seconds

    # Fallback to exponential backoff
    return min(2 ** retry_state.attempt_number, 60)


@retry(
    stop=stop_after_attempt(5),
    wait=wait_from_retry_after_header,
    retry=retry_if_exception(should_retry),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
def fetch_with_rate_limit_respect(url: str):
    """
    Fetch URL respecting rate limit headers.

    If rate limited:
    - Reads Retry-After header
    - Waits exact amount specified by server
    - Retries after wait
    """
    response = requests.get(url, timeout=10)

    if response.status_code == 429:
        retry_after = response.headers.get("Retry-After")

        if retry_after:
            if retry_after.isdigit():
                # Delay in seconds
                wait_seconds = int(retry_after)
            else:
                # HTTP date format (rare)
                retry_date = datetime.strptime(retry_after, "%a, %d %b %Y %H:%M:%S GMT")
                wait_seconds = (retry_date - datetime.now(timezone.utc)).total_seconds()
        else:
            wait_seconds = 60  # Default fallback

        raise RateLimitError(wait_seconds)

    response.raise_for_status()
    return response.json()


# Example log output:
# WARNING: Rate limited. Waiting 45s as specified by server
# INFO: Retrying in 45.0 seconds...
# INFO: Request succeeded on attempt 2

from fastapi import FastAPI, Header, HTTPException, Depends, status
from pydantic import BaseModel
import redis
import json
import hashlib

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Idempotency key TTL (24 hours)
IDEMPOTENCY_TTL = 86400


class PaymentRequest(BaseModel):
    amount: int
    currency: str
    card_token: str


def get_idempotency_key(idempotency_key: str = Header(None, alias="Idempotency-Key")):
    """Dependency to require idempotency key for non-idempotent operations"""
    if not idempotency_key:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Idempotency-Key header is required for this operation"
        )
    return idempotency_key


@app.post("/api/payments")
async def create_payment(
    payment: PaymentRequest,
    idempotency_key: str = Depends(get_idempotency_key)
):
    """
    Create a payment with idempotency protection.
    Same idempotency key returns cached result without reprocessing.
    """

    # Generate cache key
    cache_key = f"idempotency:{idempotency_key}"

    # Check if this request was already processed
    cached_result = redis_client.get(cache_key)
    if cached_result:
        # Return cached result (request was already processed)
        result = json.loads(cached_result)
        return {
            **result,
            "idempotent": True,  # Indicate this is a cached response
            "message": "Payment already processed (idempotent request)"
        }

    # Check if request is currently being processed (prevent concurrent duplicates)
    lock_key = f"lock:{cache_key}"
    if not redis_client.set(lock_key, "1", ex=30, nx=True):
        # Another request with same idempotency key is in progress
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="A request with this idempotency key is currently being processed"
        )

    try:
        # Process payment (this happens only once per idempotency key)
        result = process_payment_with_gateway(
            amount=payment.amount,
            currency=payment.currency,
            card_token=payment.card_token
        )

        # Cache the result for 24 hours
        redis_client.setex(
            cache_key,
            IDEMPOTENCY_TTL,
            json.dumps(result)
        )

        return {
            **result,
            "idempotent": False,  # First time processing this request
            "message": "Payment processed successfully"
        }

    finally:
        # Release lock
        redis_client.delete(lock_key)


def process_payment_with_gateway(amount: int, currency: str, card_token: str):
    """Simulate payment processing"""
    import uuid
    import time

    # Simulate processing time
    time.sleep(2)

    return {
        "transaction_id": str(uuid.uuid4()),
        "amount": amount,
        "currency": currency,
        "status": "completed",
        "created_at": time.time()
    }

# Python client generating idempotency keys
import uuid
import requests
from typing import Optional
import time


def create_payment(amount: int, card_token: str, api_url: str = "https://api.example.com") -> dict:
    """
    Create payment with idempotency key.

    Generates UUID v4 as idempotency key and includes it in request.
    If request fails, retries with SAME key to prevent duplicate charges.
    """
    # Generate unique idempotency key (UUID v4)
    idempotency_key = str(uuid.uuid4())
    print(f"Generated idempotency key: {idempotency_key}")

    # Store idempotency key for potential retries
    # In production, persist to database or cache
    payment_data = {
        "amount": amount,
        "currency": "USD",
        "card_token": card_token
    }

    try:
        # Make payment request with idempotency key
        response = requests.post(
            f"{api_url}/api/payments",
            json=payment_data,
            headers={
                "Content-Type": "application/json",
                "Idempotency-Key": idempotency_key,  # Include idempotency key
                "Authorization": "Bearer your_token"
            },
            timeout=10
        )

        if response.status_code == 200:
            result = response.json()

            if result.get("idempotent"):
                print("Payment already processed (idempotent retry)")
            else:
                print("Payment processed successfully")

            return result

        elif response.status_code == 409:
            # Concurrent request with same key in progress
            print("Payment is being processed, waiting...")
            time.sleep(2)
            # Retry with SAME idempotency key
            return retry_payment(idempotency_key, payment_data, api_url)

        else:
            # Other error
            response.raise_for_status()

    except requests.exceptions.RequestException as e:
        print(f"Network error: {e}, retrying with same idempotency key...")
        # On network failure, retry with SAME idempotency key
        return retry_payment(idempotency_key, payment_data, api_url)


def retry_payment(idempotency_key: str, payment_data: dict, api_url: str) -> dict:
    """
    Retry payment with the SAME idempotency key.

    This prevents duplicate charges:
    - If original request succeeded but response was lost → Returns cached result
    - If original request failed → Processes payment once
    """
    print(f"Retrying with idempotency key: {idempotency_key}")

    response = requests.post(
        f"{api_url}/api/payments",
        json=payment_data,
        headers={
            "Content-Type": "application/json",
            "Idempotency-Key": idempotency_key,  # SAME key
            "Authorization": "Bearer your_token"
        },
        timeout=10
    )

    response.raise_for_status()
    return response.json()


# Example usage
try:
    result = create_payment(
        amount=5000,  # $50.00
        card_token="tok_visa_4242"
    )
    print(f"Payment result: {result}")

except Exception as e:
    print(f"Payment failed: {e}")

import uuid
import requests
import redis
from typing import Optional

# Redis client for persistent storage
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)


class IdempotentAPIClient:
    """
    API client with persistent idempotency keys.

    Stores idempotency keys in Redis to survive:
    - Application restarts
    - Network failures
    - Client crashes
    """

    def __init__(self, api_url: str, auth_token: str):
        self.api_url = api_url
        self.auth_token = auth_token

    def _get_or_create_idempotency_key(self, operation: str, operation_id: str) -> str:
        """
        Get existing idempotency key for operation, or create new one.

        Args:
            operation: Operation type (e.g., "payment", "order")
            operation_id: Unique operation identifier (e.g., order_123)

        Returns:
            Idempotency key (UUID)
        """
        # Check if we already have a key for this operation
        cache_key = f"idempotency:{operation}:{operation_id}"
        existing_key = redis_client.get(cache_key)

        if existing_key:
            print(f"Reusing existing idempotency key: {existing_key}")
            return existing_key

        # Generate new idempotency key
        new_key = str(uuid.uuid4())
        # Store for 24 hours (same as server TTL)
        redis_client.setex(cache_key, 86400, new_key)
        print(f"Created new idempotency key: {new_key}")
        return new_key

    def _clear_idempotency_key(self, operation: str, operation_id: str):
        """Clear idempotency key after successful completion"""
        cache_key = f"idempotency:{operation}:{operation_id}"
        redis_client.delete(cache_key)
        print(f"Cleared idempotency key for {operation}:{operation_id}")

    def create_payment(self, order_id: str, amount: int, card_token: str) -> dict:
        """
        Create payment with persistent idempotency.

        If this is retried (due to failure), uses the SAME idempotency key
        that was generated for this order_id.
        """
        # Get or create idempotency key for this specific order
        idempotency_key = self._get_or_create_idempotency_key("payment", order_id)

        try:
            response = requests.post(
                f"{self.api_url}/api/payments",
                json={
                    "order_id": order_id,
                    "amount": amount,
                    "currency": "USD",
                    "card_token": card_token
                },
                headers={
                    "Content-Type": "application/json",
                    "Authorization": f"Bearer {self.auth_token}",
                    "Idempotency-Key": idempotency_key
                },
                timeout=10
            )

            response.raise_for_status()
            result = response.json()

            # Success - clear the idempotency key
            self._clear_idempotency_key("payment", order_id)

            return result

        except requests.exceptions.RequestException as e:
            # Error - keep idempotency key for retry
            print(f"Payment failed, idempotency key preserved for retry: {e}")
            raise


# Usage example
client = IdempotentAPIClient(
    api_url="https://api.example.com",
    auth_token="your_token"
)

# First attempt
try:
    result = client.create_payment(
        order_id="order_12345",
        amount=5000,
        card_token="tok_visa_4242"
    )
    print(f"Payment successful: {result}")

except Exception as e:
    print(f"Payment failed: {e}")

# If the above failed due to network error, retry:
# The SAME idempotency key will be used automatically
try:
    result = client.create_payment(
        order_id="order_12345",  # Same order ID
        amount=5000,
        card_token="tok_visa_4242"
    )
    print(f"Payment successful on retry: {result}")

except Exception as e:
    print(f"Payment failed again: {e}")


# Key benefits:
# 1. Survives application restart (keys stored in Redis)
# 2. Multiple retries use same key (no duplicates)
# 3. Different orders get different keys (isolation)
# 4. Automatic cleanup after success

from fastapi import FastAPI, Request, HTTPException, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ValidationError
from typing import Any, Optional
import logging
import traceback
import uuid

app = FastAPI()
logger = logging.getLogger(__name__)


class ErrorResponse(BaseModel):
    """Standardized error response structure"""
    error_code: str
    message: str
    status_code: int
    request_id: str
    details: Optional[Any] = None


# 1. Handle FastAPI HTTPException
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    """Handle all HTTPException instances"""
    request_id = str(uuid.uuid4())

    logger.warning(
        f"HTTPException | Request ID: {request_id} | "
        f"Status: {exc.status_code} | Detail: {exc.detail}"
    )

    error_codes = {
        400: "BAD_REQUEST",
        401: "UNAUTHORIZED",
        403: "FORBIDDEN",
        404: "NOT_FOUND",
        409: "CONFLICT",
        422: "VALIDATION_FAILED",
        429: "RATE_LIMIT_EXCEEDED",
    }

    error_code = error_codes.get(exc.status_code, "HTTP_ERROR")

    response = ErrorResponse(
        error_code=error_code,
        message=str(exc.detail),
        status_code=exc.status_code,
        request_id=request_id
    )

    return JSONResponse(
        status_code=exc.status_code,
        content=response.model_dump(exclude_none=True)
    )


# 2. Handle Pydantic validation errors (422)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Handle Pydantic validation errors with field-level details"""
    request_id = str(uuid.uuid4())

    # Format field-level errors
    errors = []
    for error in exc.errors():
        field_path = ".".join(str(loc) for loc in error['loc'][1:])
        errors.append({
            "field": field_path,
            "message": error['msg'],
            "type": error['type']
        })

    logger.warning(
        f"ValidationError | Request ID: {request_id} | "
        f"Errors: {len(errors)} | Path: {request.url.path}"
    )

    response = ErrorResponse(
        error_code="VALIDATION_FAILED",
        message=f"{len(errors)} validation error(s) occurred",
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        request_id=request_id,
        details={"errors": errors}
    )

    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=response.model_dump(exclude_none=True)
    )


# 3. Handle unexpected exceptions (500)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Catch all unexpected exceptions"""
    request_id = str(uuid.uuid4())

    # Log full error details
    logger.error(
        f"UnhandledException | Request ID: {request_id} | "
        f"Type: {type(exc).__name__} | Error: {str(exc)} | "
        f"Path: {request.url.path} | Method: {request.method}\n"
        f"Traceback: {traceback.format_exc()}"
    )

    # Production: sanitized error
    import os
    if os.getenv("ENV") == "production":
        response = ErrorResponse(
            error_code="INTERNAL_ERROR",
            message="An unexpected error occurred. Please try again later.",
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            request_id=request_id
        )
    else:
        # Development: verbose error
        response = ErrorResponse(
            error_code="INTERNAL_ERROR",
            message=str(exc),
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            request_id=request_id,
            details={
                "exception_type": type(exc).__name__,
                "traceback": traceback.format_exc().split('\n')
            }
        )

    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content=response.model_dump(exclude_none=True)
    )


# 4. Custom exceptions
class CustomAPIException(Exception):
    """Base class for custom API exceptions"""
    def __init__(
        self,
        error_code: str,
        message: str,
        status_code: int = status.HTTP_400_BAD_REQUEST,
        details: Any = None
    ):
        self.error_code = error_code
        self.message = message
        self.status_code = status_code
        self.details = details
        super().__init__(message)


@app.exception_handler(CustomAPIException)
async def custom_exception_handler(request: Request, exc: CustomAPIException):
    """Handle custom API exceptions"""
    request_id = str(uuid.uuid4())

    logger.info(
        f"CustomException | Request ID: {request_id} | "
        f"Code: {exc.error_code} | Message: {exc.message}"
    )

    response = ErrorResponse(
        error_code=exc.error_code,
        message=exc.message,
        status_code=exc.status_code,
        request_id=request_id,
        details=exc.details
    )

    return JSONResponse(
        status_code=exc.status_code,
        content=response.model_dump(exclude_none=True)
    )


# Example endpoints using the exception handlers
@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    """Example endpoint - HTTPException"""
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    return user


@app.post("/api/users")
async def create_user(username: str, email: str, age: int):
    """Example endpoint - ValidationError"""
    # Pydantic validation happens automatically
    # If validation fails, RequestValidationError is raised
    return {"username": username, "email": email, "age": age}


@app.post("/api/payments")
async def create_payment(amount: int):
    """Example endpoint - CustomAPIException"""
    if amount < 100:
        raise CustomAPIException(
            error_code="PAYMENT_AMOUNT_TOO_LOW",
            message="Minimum payment amount is $1.00 (100 cents)",
            status_code=status.HTTP_400_BAD_REQUEST,
            details={"minimum_amount": 100, "provided_amount": amount}
        )

    # Simulate unexpected error for demonstration
    if amount == 999:
        # This will be caught by global_exception_handler
        raise ValueError("Simulated unexpected error")

    return {"status": "success", "amount": amount}