Rate Limiting, Caching, and Performance

import time
import redis

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

def fixed_window_rate_limit(user_id: str, limit: int = 100, window: int = 60) -> bool:
    """
    Fixed window rate limiting.

    Args:
        user_id: User identifier
        limit: Maximum requests per window (default: 100)
        window: Time window in seconds (default: 60)

    Returns:
        True if request allowed, False if rate limited
    """
    # Get current window (timestamp rounded to window size)
    current_time = int(time.time())
    window_start = (current_time // window) * window

    # Redis key includes window start time
    key = f"rate_limit:fixed:{user_id}:{window_start}"

    # Increment counter
    count = redis_client.incr(key)

    # Set expiration on first request of window
    if count == 1:
        redis_client.expire(key, window)

    # Check if limit exceeded
    return count <= limit


# Example usage
user_id = "user_123"

# Request 1-100: Allowed
for i in range(100):
    allowed = fixed_window_rate_limit(user_id, limit=100, window=60)
    print(f"Request {i+1}: {'✓ Allowed' if allowed else '✗ Blocked'}")

# Request 101: Blocked
allowed = fixed_window_rate_limit(user_id, limit=100, window=60)
print(f"Request 101: {'✓ Allowed' if allowed else '✗ Blocked'}")  # ✗ Blocked

# After 60 seconds, counter resets
time.sleep(60)
allowed = fixed_window_rate_limit(user_id, limit=100, window=60)
print(f"Request after reset: {'✓ Allowed' if allowed else '✗ Blocked'}")  # ✓ Allowed

import time
import redis

redis_client = redis.Redis(host='localhost', port=6379)

def sliding_window_log_rate_limit(user_id: str, limit: int = 100, window: int = 60) -> tuple[bool, int]:
    """
    Sliding window log rate limiting.

    Args:
        user_id: User identifier
        limit: Maximum requests per window
        window: Time window in seconds

    Returns:
        Tuple of (allowed, remaining_requests)
    """
    key = f"rate_limit:sliding:{user_id}"
    current_time = time.time()
    window_start = current_time - window

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

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

    if request_count < limit:
        # Add current request timestamp
        redis_client.zadd(key, {str(current_time): current_time})
        # Set expiration (cleanup)
        redis_client.expire(key, window)
        return True, limit - request_count - 1
    else:
        return False, 0


# Example usage
user_id = "user_456"

# Timestamp: 10:00:00 - Send 100 requests
for i in range(100):
    allowed, remaining = sliding_window_log_rate_limit(user_id, limit=100, window=60)
    print(f"Request {i+1}: {'✓ Allowed' if allowed else '✗ Blocked'}, Remaining: {remaining}")

# Timestamp: 10:00:30 (30 seconds later) - Try to send more
time.sleep(30)
allowed, remaining = sliding_window_log_rate_limit(user_id, limit=100, window=60)
print(f"Request at +30s: {'✓ Allowed' if allowed else '✗ Blocked'}")  # ✗ Blocked

# Timestamp: 10:01:05 (65 seconds later) - Old requests expired
time.sleep(35)
allowed, remaining = sliding_window_log_rate_limit(user_id, limit=100, window=60)
print(f"Request at +65s: {'✓ Allowed' if allowed else '✗ Blocked'}")  # ✓ Allowed

import time
import redis

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

def token_bucket_rate_limit(
    user_id: str,
    capacity: int = 100,  # Max tokens in bucket
    refill_rate: float = 10,  # Tokens per second
    tokens_per_request: int = 1
) -> tuple[bool, int]:
    """
    Token bucket rate limiting.

    Args:
        user_id: User identifier
        capacity: Maximum bucket capacity (burst size)
        refill_rate: Tokens added per second
        tokens_per_request: Tokens consumed per request

    Returns:
        Tuple of (allowed, available_tokens)
    """
    key = f"rate_limit:token_bucket:{user_id}"
    current_time = time.time()

    # Get bucket state
    bucket_data = redis_client.hgetall(key)

    if not bucket_data:
        # Initialize bucket (first request)
        tokens = capacity - tokens_per_request
        redis_client.hset(key, mapping={
            "tokens": tokens,
            "last_refill": current_time
        })
        redis_client.expire(key, 3600)  # Cleanup after 1 hour
        return True, int(tokens)

    # Calculate tokens to add since last request
    last_refill = float(bucket_data["last_refill"])
    time_passed = current_time - last_refill
    tokens_to_add = time_passed * refill_rate

    # Current token count (capped at capacity)
    current_tokens = min(
        float(bucket_data["tokens"]) + tokens_to_add,
        capacity
    )

    if current_tokens >= tokens_per_request:
        # Consume tokens and allow request
        new_tokens = current_tokens - tokens_per_request
        redis_client.hset(key, mapping={
            "tokens": new_tokens,
            "last_refill": current_time
        })
        return True, int(new_tokens)
    else:
        # Not enough tokens
        return False, int(current_tokens)


# Example usage
user_id = "user_789"

# Burst: 100 rapid requests (uses all tokens)
for i in range(100):
    allowed, tokens = token_bucket_rate_limit(user_id, capacity=100, refill_rate=10)
    print(f"Request {i+1}: {'✓' if allowed else '✗'}, Tokens: {tokens}")

# Request 101: Blocked (bucket empty)
allowed, tokens = token_bucket_rate_limit(user_id, capacity=100, refill_rate=10)
print(f"Request 101: {'✓ Allowed' if allowed else '✗ Blocked'}, Tokens: {tokens}")  # ✗ Blocked

# Wait 5 seconds (50 tokens refilled at 10/sec)
time.sleep(5)
allowed, tokens = token_bucket_rate_limit(user_id, capacity=100, refill_rate=10)
print(f"After 5s: {'✓ Allowed' if allowed else '✗ Blocked'}, Tokens: {tokens}")  # ✓ Allowed (50 tokens)

# Can now make 50 more requests
for i in range(50):
    allowed, tokens = token_bucket_rate_limit(user_id, capacity=100, refill_rate=10)
    if not allowed:
        print(f"Request {i+1}: Blocked early")
        break

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

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


class RateLimiter:
    """Token bucket rate limiter with Redis"""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def check_rate_limit(
        self,
        key: str,
        limit: int,
        window: int
    ) -> tuple[bool, dict]:
        """
        Check if request should be rate limited.

        Returns:
            Tuple of (allowed, headers_dict)
        """
        current_time = time.time()
        window_start = current_time - window

        # Sliding window counter
        pipe = self.redis.pipeline()

        # Remove old entries
        pipe.zremrangebyscore(key, 0, window_start)

        # Count current window
        pipe.zcard(key)

        # Add current request
        pipe.zadd(key, {str(current_time): current_time})

        # Set expiration
        pipe.expire(key, window)

        results = pipe.execute()
        current_count = results[1]

        # Calculate remaining and reset time
        remaining = max(0, limit - current_count - 1)
        oldest_entry = self.redis.zrange(key, 0, 0, withscores=True)

        if oldest_entry:
            reset_time = int(oldest_entry[0][1] + window)
        else:
            reset_time = int(current_time + window)

        headers = {
            "X-RateLimit-Limit": str(limit),
            "X-RateLimit-Remaining": str(remaining),
            "X-RateLimit-Reset": str(reset_time),
        }

        if current_count < limit:
            return True, headers
        else:
            retry_after = reset_time - int(current_time)
            headers["Retry-After"] = str(max(0, retry_after))
            return False, headers


rate_limiter = RateLimiter(redis_client)


def rate_limit(limit: int = 100, window: int = 60):
    """
    Decorator for rate limiting endpoints.

    Args:
        limit: Maximum requests per window
        window: Time window in seconds

    Usage:
        @app.get("/api/resource")
        @rate_limit(limit=10, window=60)
        async def get_resource():
            return {"data": "..."}
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Get request object from kwargs
            request = kwargs.get('request') or args[0] if args else None

            if not request:
                # No request object, skip rate limiting
                return await func(*args, **kwargs)

            # Identify user (IP address or user ID)
            user_id = request.headers.get("X-User-ID")
            if not user_id:
                # Fallback to IP address
                user_id = request.client.host

            # Rate limit key
            key = f"rate_limit:{func.__name__}:{user_id}"

            # Check rate limit
            allowed, headers = rate_limiter.check_rate_limit(key, limit, window)

            if not allowed:
                # Rate limited
                raise HTTPException(
                    status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                    detail=f"Rate limit exceeded. Retry after {headers['Retry-After']} seconds.",
                    headers=headers
                )

            # Execute endpoint
            response = await func(*args, **kwargs)

            # Add rate limit headers to response
            if isinstance(response, dict):
                return JSONResponse(content=response, headers=headers)

            return response

        return wrapper
    return decorator


# Middleware approach (applies to all endpoints)
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    """Global rate limiting middleware"""

    # Skip rate limiting for specific paths
    if request.url.path in ["/docs", "/openapi.json", "/health"]:
        return await call_next(request)

    # Identify user
    user_id = request.headers.get("X-User-ID", request.client.host)
    key = f"rate_limit:global:{user_id}"

    # Check rate limit (1000 requests per hour)
    allowed, headers = rate_limiter.check_rate_limit(key, limit=1000, window=3600)

    if not allowed:
        return JSONResponse(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            content={
                "error": "Rate limit exceeded",
                "retry_after": headers["Retry-After"]
            },
            headers=headers
        )

    # Process request
    response = await call_next(request)

    # Add rate limit headers
    for header, value in headers.items():
        response.headers[header] = value

    return response


# Example endpoints with rate limiting
@app.get("/api/search")
@rate_limit(limit=10, window=60)  # 10 requests per minute
async def search(request: Request, q: str):
    """
    Search endpoint with strict rate limit (expensive operation).
    """
    # Perform search
    return {"query": q, "results": []}


@app.get("/api/users/{user_id}")
@rate_limit(limit=100, window=60)  # 100 requests per minute
async def get_user(request: Request, user_id: int):
    """
    Get user endpoint with standard rate limit.
    """
    return {"user_id": user_id, "username": "john_doe"}


@app.post("/api/messages")
@rate_limit(limit=5, window=60)  # 5 messages per minute
async def send_message(request: Request, to: str, message: str):
    """
    Send message endpoint with very strict rate limit (prevent spam).
    """
    return {"status": "sent", "to": to}

def get_rate_limit_key(request: Request, endpoint: str) -> str:
    """
    Generate rate limit key based on authentication status.

    Authenticated users: rate limited per user ID
    Unauthenticated: rate limited per IP address
    """
    # Check if user is authenticated
    user_id = request.headers.get("X-User-ID")  # or extract from JWT

    if user_id:
        # Per-user rate limiting
        return f"rate_limit:user:{endpoint}:{user_id}"
    else:
        # Per-IP rate limiting (stricter for anonymous users)
        ip_address = request.client.host
        return f"rate_limit:ip:{endpoint}:{ip_address}"


@app.get("/api/data")
async def get_data(request: Request):
    """Rate limit authenticated vs unauthenticated differently"""
    key = get_rate_limit_key(request, "get_data")

    # Authenticated users: 1000 req/hour
    # Anonymous users: 100 req/hour
    user_id = request.headers.get("X-User-ID")
    limit = 1000 if user_id else 100

    allowed, headers = rate_limiter.check_rate_limit(key, limit=limit, window=3600)

    if not allowed:
        raise HTTPException(status_code=429, detail="Rate limited", headers=headers)

    return {"data": "..."}

from fastapi import FastAPI, Response
from datetime import datetime, timedelta, timezone

app = FastAPI()


@app.get("/api/static-data")
async def get_static_data(response: Response):
    """
    Static data that rarely changes (cache for 1 hour).
    """
    # Set Cache-Control header
    response.headers["Cache-Control"] = "public, max-age=3600"

    # Optional: Set Expires header (fallback for HTTP/1.0)
    expires = datetime.now(timezone.utc) + timedelta(hours=1)
    response.headers["Expires"] = expires.strftime("%a, %d %b %Y %H:%M:%S GMT")

    return {"data": "This can be cached for 1 hour"}


@app.get("/api/user-profile")
async def get_user_profile(response: Response):
    """
    User-specific data (cache in browser only, 5 minutes).
    """
    # Private: only browser caches, not CDN
    response.headers["Cache-Control"] = "private, max-age=300"

    return {"user": "john_doe", "email": "john@example.com"}


@app.get("/api/realtime-data")
async def get_realtime_data(response: Response):
    """
    Real-time data that changes frequently (no caching).
    """
    # No caching
    response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate"
    response.headers["Pragma"] = "no-cache"  # HTTP/1.0 compatibility

    return {"timestamp": datetime.now(timezone.utc).isoformat(), "value": 42}


@app.get("/api/versioned-asset")
async def get_versioned_asset(response: Response):
    """
    Versioned asset (cache forever, URL changes on update).
    """
    # Cache for 1 year, immutable
    response.headers["Cache-Control"] = "public, max-age=31536000, immutable"

    return {"file": "app.abc123.js", "content": "..."}


# Result: Responses include Cache-Control headers
# Browser/CDN automatically caches according to directives

from fastapi import FastAPI, Request, Response, status
import hashlib
import json

app = FastAPI()


def generate_etag(data: dict) -> str:
    """Generate ETag from response data (MD5 hash)"""
    content = json.dumps(data, sort_keys=True)
    return hashlib.md5(content.encode()).hexdigest()


@app.get("/api/article/{article_id}")
async def get_article(article_id: int, request: Request, response: Response):
    """
    Get article with ETag support.

    Flow:
    1. Client requests article
    2. Server returns article with ETag header
    3. Client caches article + ETag
    4. Client requests article again with If-None-Match header
    5. Server compares ETags:
       - Match → 304 Not Modified (no body)
       - Different → 200 OK with new data
    """
    # Fetch article from database
    article = {
        "id": article_id,
        "title": "FastAPI Performance Guide",
        "content": "...",
        "updated_at": "2024-01-15T10:00:00Z"
    }

    # Generate ETag from article data
    etag = generate_etag(article)

    # Check if client has cached version
    client_etag = request.headers.get("If-None-Match")

    if client_etag == etag:
        # Article hasn't changed, return 304 Not Modified
        response.status_code = status.HTTP_304_NOT_MODIFIED
        response.headers["ETag"] = etag
        return None  # No body sent (saves bandwidth)
    else:
        # Article changed or first request, return full data
        response.headers["ETag"] = etag
        response.headers["Cache-Control"] = "private, must-revalidate"
        return article


# Example flow:
# Request 1: GET /api/article/123
#   Response: 200 OK, ETag: "abc123", body: {...}
#
# Request 2: GET /api/article/123
#            If-None-Match: "abc123"
#   Response: 304 Not Modified, ETag: "abc123", no body (fast!)
#
# Request 3 (after article updated): GET /api/article/123
#                                    If-None-Match: "abc123"
#   Response: 200 OK, ETag: "xyz789", body: {...} (new data)

from fastapi import FastAPI, Request, Response, status
from datetime import datetime

app = FastAPI()


@app.get("/api/document/{doc_id}")
async def get_document(doc_id: int, request: Request, response: Response):
    """Get document with Last-Modified header"""

    # Fetch document with last update time
    document = {
        "id": doc_id,
        "title": "API Documentation",
        "content": "...",
        "updated_at": datetime(2024, 1, 15, 10, 30, 0)  # From database
    }

    last_modified = document["updated_at"]

    # Check if client has cached version
    if_modified_since = request.headers.get("If-Modified-Since")

    if if_modified_since:
        # Parse client's cached timestamp
        client_time = datetime.strptime(
            if_modified_since,
            "%a, %d %b %Y %H:%M:%S GMT"
        )

        if last_modified <= client_time:
            # Document not modified since client's cache
            response.status_code = status.HTTP_304_NOT_MODIFIED
            response.headers["Last-Modified"] = last_modified.strftime(
                "%a, %d %b %Y %H:%M:%S GMT"
            )
            return None

    # Document modified or first request
    response.headers["Last-Modified"] = last_modified.strftime(
        "%a, %d %b %Y %H:%M:%S GMT"
    )
    response.headers["Cache-Control"] = "private, must-revalidate"

    return document


# Example flow:
# Request 1: GET /api/document/456
#   Response: 200 OK
#             Last-Modified: Mon, 15 Jan 2024 10:30:00 GMT
#             Body: {...}
#
# Request 2: GET /api/document/456
#            If-Modified-Since: Mon, 15 Jan 2024 10:30:00 GMT
#   Response: 304 Not Modified (document unchanged)
#
# Request 3 (after update): GET /api/document/456
#                          If-Modified-Since: Mon, 15 Jan 2024 10:30:00 GMT
#   Response: 200 OK
#             Last-Modified: Mon, 15 Jan 2024 14:20:00 GMT
#             Body: {...} (new data)

from fastapi import FastAPI
import redis
import json
from functools import wraps
import hashlib

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


def cache_result(ttl: int = 300):
    """
    Decorator to cache function results in Redis.

    Args:
        ttl: Time to live in seconds (default: 5 minutes)

    Usage:
        @cache_result(ttl=3600)
        async def expensive_query(user_id: int):
            return database.query(...)
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Generate cache key from function name and arguments
            key_data = f"{func.__name__}:{args}:{kwargs}"
            cache_key = f"cache:{hashlib.md5(key_data.encode()).hexdigest()}"

            # Try to get from cache
            cached = redis_client.get(cache_key)
            if cached:
                # Cache hit!
                return json.loads(cached)

            # Cache miss, execute function
            result = await func(*args, **kwargs)

            # Store in cache
            redis_client.setex(
                cache_key,
                ttl,
                json.dumps(result, default=str)  # Handle datetime, etc.
            )

            return result

        return wrapper
    return decorator


@app.get("/api/stats/daily")
@cache_result(ttl=3600)  # Cache for 1 hour
async def get_daily_stats():
    """
    Get daily statistics (expensive query, cached).

    First request: queries database (slow)
    Subsequent requests: returns cached result (fast)
    """
    # Simulate expensive database query
    import time
    time.sleep(2)  # 2 seconds to compute

    stats = {
        "date": "2024-01-15",
        "users": 1523,
        "revenue": 45678.90,
        "requests": 98765
    }

    return stats


# Request 1: Takes 2 seconds (database query)
# Requests 2-N (within 1 hour): Takes 5ms (Redis cache)


@app.get("/api/user/{user_id}/dashboard")
@cache_result(ttl=300)  # Cache for 5 minutes
async def get_user_dashboard(user_id: int):
    """
    Get user dashboard (personalized, cached per user).
    """
    # Complex aggregation query
    dashboard = {
        "user_id": user_id,
        "total_tasks": 42,
        "completed_tasks": 35,
        "pending_tasks": 7,
        "recent_activity": [...]
    }

    return dashboard


# Each user has separate cache entry
# user/123/dashboard → cached separately from user/456/dashboard

import redis
import json

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


def cache_with_ttl(key: str, data: dict, ttl: int):
    """
    Cache with automatic expiration.

    Args:
        key: Cache key
        data: Data to cache
        ttl: Time to live in seconds

    Example TTLs:
    - 60s: Real-time data (stock prices)
    - 300s (5 min): User dashboards
    - 3600s (1 hour): Product listings
    - 86400s (24 hours): Static content
    """
    redis_client.setex(key, ttl, json.dumps(data))


# Example usage
cache_with_ttl(
    key="product:123",
    data={"id": 123, "name": "Widget", "price": 19.99},
    ttl=3600  # Cache for 1 hour
)

# After 1 hour, Redis automatically deletes the key
# Next request will be a cache miss and fetch from database

from fastapi import FastAPI
import redis

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


def invalidate_user_cache(user_id: int):
    """
    Invalidate all cache entries related to a user.

    Called when:
    - User profile updated
    - User permissions changed
    - User deleted
    """
    # Pattern match and delete all user-related cache keys
    pattern = f"cache:user:{user_id}:*"
    keys = redis_client.keys(pattern)

    if keys:
        deleted = redis_client.delete(*keys)
        print(f"Invalidated {deleted} cache entries for user {user_id}")


@app.put("/api/users/{user_id}")
async def update_user(user_id: int, name: str, email: str):
    """
    Update user and invalidate cache.

    Flow:
    1. Update database
    2. Invalidate all related caches
    3. Next request will fetch fresh data from database
    """
    # Update database
    database.update_user(user_id, name=name, email=email)

    # Invalidate cached user data
    invalidate_user_cache(user_id)

    return {"message": "User updated", "cache_invalidated": True}


# Example cache keys that get invalidated:
# - cache:user:123:profile
# - cache:user:123:settings
# - cache:user:123:permissions
# All deleted in one operation!

def get_with_cache_aside(key: str, fetch_func, ttl: int = 300):
    """
    Cache-aside pattern (also called lazy loading).

    Flow:
    1. Check cache
    2. If hit → return cached data (fast!)
    3. If miss → fetch from database (slow)
    4. Store in cache for next time
    5. Return data

    Args:
        key: Cache key
        fetch_func: Function to fetch data on cache miss
        ttl: Time to live in seconds
    """
    # Try cache first
    cached = redis_client.get(key)
    if cached:
        print(f"Cache HIT for {key}")
        return json.loads(cached)

    print(f"Cache MISS for {key}, fetching from source...")

    # Cache miss - fetch from source
    data = fetch_func()

    # Store in cache for next time
    redis_client.setex(key, ttl, json.dumps(data))
    print(f"Cached {key} for {ttl}s")

    return data


# Example usage
def fetch_product_from_db(product_id: int):
    """Slow database query"""
    return database.query(Product).filter(Product.id == product_id).first()


@app.get("/api/products/{product_id}")
async def get_product(product_id: int):
    """Get product with cache-aside pattern"""

    product = get_with_cache_aside(
        key=f"product:{product_id}",
        fetch_func=lambda: fetch_product_from_db(product_id),
        ttl=3600  # Cache for 1 hour
    )

    return product


# Log output:
# Request 1: Cache MISS for product:123, fetching from source... (150ms)
# Request 2: Cache HIT for product:123 (5ms)
# Request 3: Cache HIT for product:123 (5ms)

def update_with_write_through(user_id: int, data: dict):
    """
    Write-through pattern.

    Flow:
    1. Update database
    2. Update cache immediately
    3. Return success

    Pros:
    - Cache always fresh (never stale)
    - Next read is guaranteed cache hit

    Cons:
    - Extra write operations (slower writes)
    - Caches data that might never be read
    """
    # Update database (slow)
    database.update_user(user_id, data)

    # Update cache immediately (fast)
    cache_key = f"cache:user:{user_id}"
    redis_client.setex(cache_key, 3600, json.dumps(data))

    print(f"Updated database and cache for user {user_id}")


@app.put("/api/users/{user_id}/profile")
async def update_user_profile(user_id: int, name: str, bio: str):
    """Update user profile with write-through caching"""

    data = {"id": user_id, "name": name, "bio": bio}

    # Update both database AND cache
    update_with_write_through(user_id, data)

    return {"message": "Profile updated", "cache_updated": True}


# Timeline:
# 1. Client: PUT /api/users/123/profile
# 2. Server: Update database (100ms)
# 3. Server: Update Redis cache (2ms)
# 4. Server: Return success
# 5. Next GET request: Cache hit! (5ms, fresh data)

def set_with_tags(key: str, data: dict, tags: list[str], ttl: int):
    """
    Tag cache entries for batch invalidation.

    Args:
        key: Cache key
        data: Data to cache
        tags: List of tags for grouping
        ttl: Time to live

    Example:
        Product 123 in category "widgets" and brand "acme"
        Tags: ["products", "category:widgets", "brand:acme"]

        When category updated: invalidate all "category:widgets"
        When brand updated: invalidate all "brand:acme"
    """
    # Store data
    redis_client.setex(key, ttl, json.dumps(data))

    # Store tags (track which keys have which tags)
    for tag in tags:
        redis_client.sadd(f"cache_tag:{tag}", key)
        redis_client.expire(f"cache_tag:{tag}", ttl)


def invalidate_by_tag(tag: str):
    """
    Invalidate all caches with specific tag.

    Example:
        invalidate_by_tag("category:widgets")
        → Invalidates all products in widgets category
    """
    tag_key = f"cache_tag:{tag}"
    cache_keys = redis_client.smembers(tag_key)

    if cache_keys:
        # Delete all tagged cache entries
        deleted = redis_client.delete(*cache_keys)
        # Delete tag set itself
        redis_client.delete(tag_key)

        print(f"Invalidated {deleted} cache entries with tag '{tag}'")


# Usage example
@app.post("/api/products")
async def create_product(name: str, category: str, brand: str):
    """Create product and cache it with tags"""

    product = {"id": 123, "name": name, "category": category, "brand": brand}

    # Cache with multiple tags
    set_with_tags(
        key="cache:product:123",
        data=product,
        tags=["products", f"category:{category}", f"brand:{brand}"],
        ttl=3600
    )

    return product


@app.put("/api/categories/{category}/discount")
async def update_category_discount(category: str, discount: float):
    """
    Update category discount - invalidates ALL products in category.

    Example: Category "widgets" gets 20% discount
    → Invalidate all cached widget products
    → Next requests fetch products with new discount applied
    """
    database.update_category_discount(category, discount)

    # Invalidate all products in this category
    invalidate_by_tag(f"category:{category}")

    return {"message": f"Updated {category} discount, cache invalidated"}

from sqlalchemy.orm import Session
from fastapi import Depends
import redis
import json

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


async def get_user_with_cache(user_id: int, db: Session) -> dict:
    """
    Get user from cache or database.

    Flow:
    1. Check Redis cache
    2. If cache hit → return immediately (fast)
    3. If cache miss → query database (slow)
    4. Store result in cache
    5. Return data
    """
    cache_key = f"user:{user_id}"

    # Try cache first
    cached_user = redis_client.get(cache_key)
    if cached_user:
        print(f"Cache HIT for user {user_id}")
        return json.loads(cached_user)

    print(f"Cache MISS for user {user_id}, querying database...")

    # Cache miss - query database
    user = db.query(User).filter(User.id == user_id).first()

    if not user:
        return None

    # Serialize user data
    user_data = {
        "id": user.id,
        "username": user.username,
        "email": user.email,
        "created_at": user.created_at.isoformat()
    }

    # Store in cache (5 minutes TTL)
    redis_client.setex(cache_key, 300, json.dumps(user_data))

    return user_data


@app.get("/api/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
    """Get user (cached)"""
    user = await get_user_with_cache(user_id, db)

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return user


# Performance:
# First request (user 123): 150ms (database query)
# Second request (user 123): 5ms (Redis cache)
# 30× faster!

from fastapi import FastAPI, Response

app = FastAPI()


@app.get("/api/products")
async def get_products(response: Response):
    """
    Public product list (cacheable at CDN).

    Cache-Control directives:
    - public: Allow CDN caching
    - max-age=3600: Browser cache for 1 hour
    - s-maxage=86400: CDN cache for 24 hours
    - stale-while-revalidate=300: Serve stale for 5 min while refreshing
    """
    response.headers["Cache-Control"] = "public, max-age=3600, s-maxage=86400, stale-while-revalidate=300"

    # Add Vary header for content negotiation
    response.headers["Vary"] = "Accept-Encoding, Accept-Language"

    products = [
        {"id": 1, "name": "Widget", "price": 19.99},
        {"id": 2, "name": "Gadget", "price": 29.99}
    ]

    return {"products": products}


@app.get("/api/user/profile")
async def get_user_profile(response: Response, user_id: str):
    """
    User-specific data (not cacheable at CDN).

    Cache-Control:
    - private: Only browser caches, NOT CDN
    - max-age=300: Cache in browser for 5 minutes
    """
    response.headers["Cache-Control"] = "private, max-age=300"

    return {"user_id": user_id, "name": "John Doe"}


@app.get("/api/realtime/stock/{symbol}")
async def get_stock_price(symbol: str, response: Response):
    """
    Real-time data (no caching).

    Cache-Control:
    - no-store: Never cache anywhere
    - must-revalidate: Always check with origin
    """
    response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate"

    return {"symbol": symbol, "price": 150.25, "timestamp": "2024-01-15T10:30:00Z"}

import requests

# Example: Cloudflare Cache Purge
def purge_cloudflare_cache(zone_id: str, api_token: str, urls: list[str]):
    """
    Purge specific URLs from Cloudflare CDN cache.

    Use cases:
    - Product updated → purge /api/products and /api/products/{id}
    - User data changed → purge /api/users/{user_id}
    """
    headers = {
        "Authorization": f"Bearer {api_token}",
        "Content-Type": "application/json"
    }

    payload = {
        "files": urls
    }

    response = requests.post(
        f"https://api.cloudflare.com/client/v4/zones/{zone_id}/purge_cache",
        json=payload,
        headers=headers
    )

    return response.json()


# Example: Purge when product updated
@app.put("/api/products/{product_id}")
async def update_product(product_id: int, name: str, price: float):
    """Update product and purge CDN cache"""

    # Update database
    database.update_product(product_id, name=name, price=price)

    # Purge CDN cache for affected URLs
    purge_cloudflare_cache(
        zone_id="your_zone_id",
        api_token="your_api_token",
        urls=[
            f"https://api.example.com/api/products",
            f"https://api.example.com/api/products/{product_id}"
        ]
    )

    return {"message": "Product updated and cache purged"}


# Alternative: Cache-Tag-based purging (Fastly, CloudFront)
@app.get("/api/products/{product_id}")
async def get_product(product_id: int, response: Response):
    """Return product with cache tags for efficient purging"""

    product = database.get_product(product_id)

    # Add cache tags (Fastly Surrogate-Key header)
    response.headers["Surrogate-Key"] = f"product product-{product_id} category-{product.category_id}"
    response.headers["Cache-Control"] = "public, max-age=3600, s-maxage=86400"

    return product

# Purge all products in category 5:
# fastly.purge_key("category-5")  # Purges all products in category 5

from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

# Add gzip compression middleware
app.add_middleware(
    GZipMiddleware,
    minimum_size=1000  # Only compress responses > 1KB
)


@app.get("/api/large-dataset")
async def get_large_dataset():
    """
    Large JSON response (automatically compressed).

    Without compression: 250 KB
    With gzip: 35 KB (86% reduction!)
    """
    dataset = [
        {
            "id": i,
            "name": f"Item {i}",
            "description": "A long description " * 10,
            "tags": ["tag1", "tag2", "tag3"],
            "metadata": {"key": "value"}
        }
        for i in range(1000)
    ]

    return {"data": dataset, "total": len(dataset)}


# Headers sent by client:
# Accept-Encoding: gzip, deflate, br
#
# Headers sent by server:
# Content-Encoding: gzip
# Content-Length: 35000  (compressed size)
# Vary: Accept-Encoding

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
import brotli
import json
from functools import wraps

app = FastAPI()


class BrotliMiddleware:
    """Custom Brotli compression middleware"""

    def __init__(self, app, minimum_size: int = 1000, quality: int = 5):
        self.app = app
        self.minimum_size = minimum_size
        self.quality = quality  # 0-11 (higher = better compression, slower)

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        # Check if client accepts brotli
        headers = dict(scope.get("headers", []))
        accept_encoding = headers.get(b"accept-encoding", b"").decode()

        if "br" not in accept_encoding:
            # Client doesn't support brotli
            await self.app(scope, receive, send)
            return

        # Intercept response
        response_body = []

        async def send_wrapper(message):
            if message["type"] == "http.response.body":
                response_body.append(message.get("body", b""))
            else:
                await send(message)

        await self.app(scope, receive, send_wrapper)

        # Compress response
        body = b"".join(response_body)

        if len(body) >= self.minimum_size:
            compressed = brotli.compress(body, quality=self.quality)

            await send({
                "type": "http.response.body",
                "body": compressed,
                "headers": [
                    (b"content-encoding", b"br"),
                    (b"vary", b"Accept-Encoding"),
                    (b"content-length", str(len(compressed)).encode())
                ]
            })
        else:
            await send({
                "type": "http.response.body",
                "body": body
            })


# Add Brotli middleware
app.add_middleware(BrotliMiddleware, minimum_size=1000, quality=5)


@app.get("/api/products")
async def get_products():
    """
    Returns large product list (automatically brotli compressed).

    Original size: 180 KB
    Gzip: 28 KB
    Brotli (quality=5): 22 KB (12% better than gzip!)
    """
    products = [{"id": i, "name": f"Product {i}", "price": 19.99 + i} for i in range(5000)]
    return {"products": products}

from fastapi import FastAPI, Request, Response
import gzip
import brotli
import json

app = FastAPI()


def compress_response(body: bytes, accept_encoding: str) -> tuple[bytes, str]:
    """
    Choose best compression based on client support.

    Priority: brotli > gzip > none
    """
    encodings = accept_encoding.lower().split(",")
    encodings = [e.strip() for e in encodings]

    # Prefer brotli (best compression)
    if "br" in encodings:
        compressed = brotli.compress(body, quality=5)
        return compressed, "br"

    # Fallback to gzip
    elif "gzip" in encodings:
        compressed = gzip.compress(body)
        return compressed, "gzip"

    # No compression supported
    else:
        return body, "identity"


@app.get("/api/data")
async def get_data(request: Request, response: Response):
    """Compress response based on client capabilities"""

    data = {"items": [{"id": i, "value": f"Item {i}"} for i in range(1000)]}
    body = json.dumps(data).encode()

    # Check client's Accept-Encoding header
    accept_encoding = request.headers.get("accept-encoding", "")

    compressed_body, encoding = compress_response(body, accept_encoding)

    # Set headers
    response.headers["Content-Encoding"] = encoding
    response.headers["Vary"] = "Accept-Encoding"
    response.headers["Content-Length"] = str(len(compressed_body))

    return Response(content=compressed_body, media_type="application/json")

from fastapi import FastAPI, Request
import time
import logging

app = FastAPI()

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


@app.middleware("http")
async def log_request_time(request: Request, call_next):
    """
    Middleware to measure and log request processing time.

    Logs:
    - Endpoint path
    - HTTP method
    - Status code
    - Processing time
    - Client IP
    """
    start_time = time.time()

    # Process request
    response = await call_next(request)

    # Calculate duration
    duration = time.time() - start_time
    duration_ms = round(duration * 1000, 2)

    # Add timing header to response
    response.headers["X-Process-Time"] = f"{duration_ms}ms"

    # Log request details
    logger.info(
        f"{request.method} {request.url.path} "
        f"status={response.status_code} "
        f"duration={duration_ms}ms "
        f"client={request.client.host}"
    )

    # Alert on slow requests
    if duration_ms > 1000:
        logger.warning(f"Slow request: {request.url.path} took {duration_ms}ms")

    return response


@app.get("/api/fast")
async def fast_endpoint():
    """Fast endpoint"""
    return {"message": "Quick response"}

# Log output:
# INFO: GET /api/fast status=200 duration=5.23ms client=127.0.0.1
# Response includes: X-Process-Time: 5.23ms


@app.get("/api/slow")
async def slow_endpoint():
    """Slow endpoint (triggers warning)"""
    time.sleep(1.5)
    return {"message": "Slow response"}

# Log output:
# WARNING: Slow request: /api/slow took 1502.45ms
# INFO: GET /api/slow status=200 duration=1502.45ms client=127.0.0.1

from fastapi import FastAPI, Request
from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
from fastapi.responses import Response
import time

app = FastAPI()

# Define metrics
REQUEST_COUNT = Counter(
    'api_requests_total',
    'Total API requests',
    ['method', 'endpoint', 'status']
)

REQUEST_DURATION = Histogram(
    'api_request_duration_seconds',
    'Request duration in seconds',
    ['method', 'endpoint'],
    buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)

ACTIVE_REQUESTS = Gauge(
    'api_requests_active',
    'Number of requests currently being processed'
)

CACHE_HITS = Counter('cache_hits_total', 'Total cache hits')
CACHE_MISSES = Counter('cache_misses_total', 'Total cache misses')

DATABASE_QUERY_DURATION = Histogram(
    'database_query_duration_seconds',
    'Database query duration',
    ['query_type'],
    buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0]
)


@app.middleware("http")
async def prometheus_middleware(request: Request, call_next):
    """Collect metrics for every request"""

    # Track active requests
    ACTIVE_REQUESTS.inc()

    # Start timer
    start_time = time.time()

    try:
        # Process request
        response = await call_next(request)

        # Record duration
        duration = time.time() - start_time
        REQUEST_DURATION.labels(
            method=request.method,
            endpoint=request.url.path
        ).observe(duration)

        # Count request
        REQUEST_COUNT.labels(
            method=request.method,
            endpoint=request.url.path,
            status=response.status_code
        ).inc()

        return response

    finally:
        # Decrement active requests
        ACTIVE_REQUESTS.dec()


@app.get("/metrics")
async def metrics():
    """
    Prometheus metrics endpoint.

    Scrape this endpoint with Prometheus:
    scrape_configs:
      - job_name: 'my-api'
        static_configs:
          - targets: ['localhost:8000']
    """
    return Response(
        content=generate_latest(),
        media_type=CONTENT_TYPE_LATEST
    )


@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    """Example endpoint with cache metrics"""

    # Check cache
    cached = redis_client.get(f"user:{user_id}")

    if cached:
        CACHE_HITS.inc()
        return json.loads(cached)
    else:
        CACHE_MISSES.inc()

        # Track database query time
        query_start = time.time()
        user = database.get_user(user_id)
        query_duration = time.time() - query_start

        DATABASE_QUERY_DURATION.labels(query_type="get_user").observe(query_duration)

        # Cache result
        redis_client.setex(f"user:{user_id}", 300, json.dumps(user))

        return user


# Example Prometheus queries (PromQL):
#
# Request rate:
#   rate(api_requests_total[5m])
#
# Error rate:
#   rate(api_requests_total{status=~"5.."}[5m])
#
# P95 latency:
#   histogram_quantile(0.95, rate(api_request_duration_seconds_bucket[5m]))
#
# Cache hit rate:
#   rate(cache_hits_total[5m]) / (rate(cache_hits_total[5m]) + rate(cache_misses_total[5m]))

import random
from locust import HttpUser, task, between

class APIUser(HttpUser):
    """
    Simulate API user behavior.

    Run test:
    locust -f locustfile.py --host=http://localhost:8000

    Open http://localhost:8089 to start load test
    """

    # Wait 1-3 seconds between requests
    wait_time = between(1, 3)

    @task(10)  # Weight: 10 (most common)
    def get_products(self):
        """Get product list (cacheable, should be fast)"""
        self.client.get("/api/products")

    @task(5)  # Weight: 5
    def get_user(self):
        """Get user details"""
        user_id = random.randint(1, 1000)
        self.client.get(f"/api/users/{user_id}")

    @task(2)  # Weight: 2
    def search(self):
        """Search (expensive, rate limited)"""
        query = random.choice(["python", "fastapi", "api"])
        self.client.get(f"/api/search?q={query}")

    @task(1)  # Weight: 1 (least common)
    def create_order(self):
        """Create order (write operation)"""
        self.client.post("/api/orders", json={
            "product_id": random.randint(1, 100),
            "quantity": random.randint(1, 5)
        })


# Load test results example:
# Name                   # Reqs    # Fails    Avg      Min      Max     Median   RPS
# GET /api/products      10000     0          15ms     8ms      250ms   12ms     100
# GET /api/users/:id     5000      0          25ms     10ms     500ms   20ms     50
# GET /api/search        2000      15         450ms    200ms    2000ms  400ms    20
# POST /api/orders       1000      0          80ms     50ms     800ms   75ms     10
#
# Analysis:
# - Products endpoint: Fast (cached), handles 100 RPS easily
# - Search endpoint: Slow (expensive), 15 failures (rate limited)
# - Need to optimize search or increase rate limit