API Design

from fastapi import FastAPI, HTTPException, Query, Path, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel, EmailStr, Field
from typing import Optional, List, Generic, TypeVar
from datetime import datetime
from uuid import uuid4

T = TypeVar('T')

# ============================================
# Response Models
# ============================================

class MetaInfo(BaseModel):
    request_id: str
    timestamp: datetime = Field(default_factory=datetime.utcnow)

class PaginationInfo(BaseModel):
    page: int
    per_page: int
    total: int
    total_pages: int
    next_cursor: Optional[str] = None
    prev_cursor: Optional[str] = None

class ApiResponse(BaseModel, Generic[T]):
    success: bool = True
    data: T
    meta: MetaInfo

class PaginatedResponse(BaseModel, Generic[T]):
    success: bool = True
    data: List[T]
    pagination: PaginationInfo
    meta: MetaInfo

class ErrorDetail(BaseModel):
    field: Optional[str] = None
    message: str
    code: str

class ErrorResponse(BaseModel):
    success: bool = False
    error: dict
    meta: MetaInfo

# ============================================
# Domain Models
# ============================================

class UserBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    email: Optional[EmailStr] = None

class User(UserBase):
    id: int
    created_at: datetime
    updated_at: datetime
    
    class Config:
        from_attributes = True

# ============================================
# API Implementation
# ============================================

app = FastAPI(title="User API", version="1.0.0")

# Request ID middleware
@app.middleware("http")
async def add_request_id(request, call_next):
    request_id = str(uuid4())
    request.state.request_id = request_id
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

# Custom exception handler
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(
            error={
                "code": exc.detail.get("code", "ERROR"),
                "message": exc.detail.get("message", str(exc.detail)),
                "details": exc.detail.get("details", [])
            },
            meta=MetaInfo(request_id=request.state.request_id)
        ).dict()
    )

# List users with pagination
@app.get("/users", response_model=PaginatedResponse[User])
async def list_users(
    request: Request,
    page: int = Query(1, ge=1),
    per_page: int = Query(20, ge=1, le=100),
    sort: str = Query("created_at", regex="^-?(name|email|created_at)$"),
    search: Optional[str] = Query(None, min_length=1)
):
    # Parse sort direction
    sort_desc = sort.startswith("-")
    sort_field = sort.lstrip("-")
    
    # Build query (pseudo-code)
    query = db.query(UserModel)
    if search:
        query = query.filter(UserModel.name.ilike(f"%{search}%"))
    
    total = query.count()
    users = query.order_by(
        getattr(UserModel, sort_field).desc() if sort_desc else getattr(UserModel, sort_field)
    ).offset((page - 1) * per_page).limit(per_page).all()
    
    return PaginatedResponse(
        data=users,
        pagination=PaginationInfo(
            page=page,
            per_page=per_page,
            total=total,
            total_pages=(total + per_page - 1) // per_page
        ),
        meta=MetaInfo(request_id=request.state.request_id)
    )

# Get single user
@app.get("/users/{user_id}", response_model=ApiResponse[User])
async def get_user(
    request: Request,
    user_id: int = Path(..., ge=1)
):
    user = db.query(UserModel).filter(UserModel.id == user_id).first()
    
    if not user:
        raise HTTPException(
            status_code=404,
            detail={
                "code": "USER_NOT_FOUND",
                "message": f"User with id {user_id} not found"
            }
        )
    
    return ApiResponse(
        data=user,
        meta=MetaInfo(request_id=request.state.request_id)
    )

# Create user
@app.post("/users", response_model=ApiResponse[User], status_code=201)
async def create_user(
    request: Request,
    user_data: UserCreate
):
    # Check for existing email
    existing = db.query(UserModel).filter(UserModel.email == user_data.email).first()
    if existing:
        raise HTTPException(
            status_code=400,
            detail={
                "code": "VALIDATION_ERROR",
                "message": "Email already registered",
                "details": [{"field": "email", "message": "This email is already in use"}]
            }
        )
    
    user = UserModel(**user_data.dict())
    db.add(user)
    db.commit()
    db.refresh(user)
    
    return ApiResponse(
        data=user,
        meta=MetaInfo(request_id=request.state.request_id)
    )

# Update user (partial)
@app.patch("/users/{user_id}", response_model=ApiResponse[User])
async def update_user(
    request: Request,
    user_id: int = Path(..., ge=1),
    user_data: UserUpdate = None
):
    user = db.query(UserModel).filter(UserModel.id == user_id).first()
    
    if not user:
        raise HTTPException(status_code=404, detail={"code": "USER_NOT_FOUND", "message": "User not found"})
    
    update_data = user_data.dict(exclude_unset=True)
    for field, value in update_data.items():
        setattr(user, field, value)
    
    user.updated_at = datetime.utcnow()
    db.commit()
    db.refresh(user)
    
    return ApiResponse(
        data=user,
        meta=MetaInfo(request_id=request.state.request_id)
    )

# Delete user
@app.delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: int = Path(..., ge=1)):
    user = db.query(UserModel).filter(UserModel.id == user_id).first()
    
    if not user:
        raise HTTPException(status_code=404, detail={"code": "USER_NOT_FOUND", "message": "User not found"})
    
    db.delete(user)
    db.commit()
    
    return None  # 204 No Content

import strawberry
from strawberry.types import Info
from strawberry.dataloader import DataLoader
from typing import List, Optional
from datetime import datetime
import asyncio

# ============================================
# DataLoaders (Solve N+1 Problem)
# ============================================

async def load_users(keys: List[int]) -> List["UserType"]:
    """Batch load users by IDs"""
    users = await User.filter(id__in=keys)
    user_map = {u.id: u for u in users}
    return [user_map.get(key) for key in keys]

async def load_posts_by_author(keys: List[int]) -> List[List["PostType"]]:
    """Batch load posts for multiple authors"""
    posts = await Post.filter(author_id__in=keys)
    posts_by_author = {}
    for post in posts:
        if post.author_id not in posts_by_author:
            posts_by_author[post.author_id] = []
        posts_by_author[post.author_id].append(post)
    return [posts_by_author.get(key, []) for key in keys]

async def load_followers_count(keys: List[int]) -> List[int]:
    """Batch load follower counts"""
    counts = await Follow.filter(following_id__in=keys).values('following_id').annotate(count=Count('id'))
    count_map = {c['following_id']: c['count'] for c in counts}
    return [count_map.get(key, 0) for key in keys]


def get_dataloaders() -> dict:
    return {
        "user_loader": DataLoader(load_fn=load_users),
        "posts_loader": DataLoader(load_fn=load_posts_by_author),
        "followers_count_loader": DataLoader(load_fn=load_followers_count)
    }


# ============================================
# Types
# ============================================

@strawberry.type
class UserType:
    id: strawberry.ID
    name: str
    email: str
    created_at: datetime
    
    @strawberry.field
    async def posts(self, info: Info) -> List["PostType"]:
        """Lazily loaded posts using DataLoader"""
        loader = info.context["posts_loader"]
        return await loader.load(self.id)
    
    @strawberry.field
    async def followers_count(self, info: Info) -> int:
        """Efficient batch-loaded follower count"""
        loader = info.context["followers_count_loader"]
        return await loader.load(self.id)
    
    @strawberry.field
    async def followers(
        self, 
        info: Info,
        limit: int = 10,
        offset: int = 0
    ) -> List["UserType"]:
        """Paginated followers list"""
        follows = await Follow.filter(following_id=self.id).offset(offset).limit(limit)
        follower_ids = [f.follower_id for f in follows]
        loader = info.context["user_loader"]
        return await asyncio.gather(*[loader.load(id) for id in follower_ids])


@strawberry.type
class PostType:
    id: strawberry.ID
    title: str
    content: str
    author_id: strawberry.Private[int]  # Hidden from schema
    likes: int
    created_at: datetime
    
    @strawberry.field
    async def author(self, info: Info) -> UserType:
        loader = info.context["user_loader"]
        return await loader.load(self.author_id)
    
    @strawberry.field
    async def comments(
        self,
        info: Info,
        limit: int = 20
    ) -> List["CommentType"]:
        return await Comment.filter(post_id=self.id).limit(limit)


# ============================================
# Inputs
# ============================================

@strawberry.input
class CreateUserInput:
    name: str
    email: str
    password: str

@strawberry.input
class UpdateUserInput:
    name: Optional[str] = None
    email: Optional[str] = None

@strawberry.input
class CreatePostInput:
    title: str
    content: str


# ============================================
# Query & Mutation
# ============================================

@strawberry.type
class Query:
    @strawberry.field
    async def user(self, info: Info, id: strawberry.ID) -> Optional[UserType]:
        loader = info.context["user_loader"]
        return await loader.load(int(id))
    
    @strawberry.field
    async def users(
        self,
        limit: int = 20,
        offset: int = 0,
        search: Optional[str] = None
    ) -> List[UserType]:
        query = User.all()
        if search:
            query = query.filter(name__icontains=search)
        return await query.offset(offset).limit(limit)
    
    @strawberry.field
    async def feed(
        self,
        user_id: strawberry.ID,
        limit: int = 20,
        cursor: Optional[str] = None
    ) -> List[PostType]:
        """User's feed with cursor pagination"""
        query = Post.filter(author_id=int(user_id))
        
        if cursor:
            # Decode cursor (base64 encoded post ID)
            last_id = int(base64.b64decode(cursor).decode())
            query = query.filter(id__lt=last_id)
        
        return await query.order_by("-created_at").limit(limit)


@strawberry.type
class Mutation:
    @strawberry.mutation
    async def create_user(self, input: CreateUserInput) -> UserType:
        # Validate email uniqueness
        existing = await User.filter(email=input.email).first()
        if existing:
            raise ValueError("Email already registered")
        
        user = await User.create(
            name=input.name,
            email=input.email,
            password=hash_password(input.password)
        )
        return user
    
    @strawberry.mutation
    async def update_user(
        self,
        id: strawberry.ID,
        input: UpdateUserInput,
        info: Info
    ) -> UserType:
        # Authorization check
        current_user = info.context["current_user"]
        if current_user.id != int(id):
            raise PermissionError("Not authorized")
        
        update_data = {k: v for k, v in input.__dict__.items() if v is not None}
        await User.filter(id=int(id)).update(**update_data)
        
        return await User.get(id=int(id))
    
    @strawberry.mutation
    async def create_post(
        self,
        input: CreatePostInput,
        info: Info
    ) -> PostType:
        current_user = info.context["current_user"]
        
        post = await Post.create(
            title=input.title,
            content=input.content,
            author_id=current_user.id
        )
        
        # Publish to subscribers
        await info.context["pubsub"].publish(
            f"post_created:{current_user.id}",
            post
        )
        
        return post


# ============================================
# Subscription
# ============================================

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def post_created(
        self,
        info: Info,
        user_id: strawberry.ID
    ) -> PostType:
        async for post in info.context["pubsub"].subscribe(f"post_created:{user_id}"):
            yield post


# ============================================
# Schema Setup
# ============================================

schema = strawberry.Schema(
    query=Query,
    mutation=Mutation,
    subscription=Subscription
)

# FastAPI integration
from strawberry.fastapi import GraphQLRouter

async def get_context():
    return {
        **get_dataloaders(),
        "current_user": get_current_user(),
        "pubsub": pubsub
    }

graphql_app = GraphQLRouter(schema, context_getter=get_context)
app.include_router(graphql_app, prefix="/graphql")

import redis.asyncio as redis
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.responses import JSONResponse
from functools import wraps
from typing import Optional, Callable
from dataclasses import dataclass
from enum import Enum
import time
import hashlib

class RateLimitAlgorithm(Enum):
    FIXED_WINDOW = "fixed_window"
    SLIDING_WINDOW = "sliding_window"
    TOKEN_BUCKET = "token_bucket"

@dataclass
class RateLimitConfig:
    requests: int
    window_seconds: int
    algorithm: RateLimitAlgorithm = RateLimitAlgorithm.SLIDING_WINDOW
    key_prefix: str = "ratelimit"

class RateLimiter:
    """Production-ready distributed rate limiter using Redis"""
    
    # Lua script for atomic sliding window
    SLIDING_WINDOW_SCRIPT = """
    local key = KEYS[1]
    local now = tonumber(ARGV[1])
    local window = tonumber(ARGV[2])
    local limit = tonumber(ARGV[3])
    
    -- Remove old entries
    redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
    
    -- Count current entries
    local current = redis.call('ZCARD', key)
    
    if current < limit then
        -- Add new request
        redis.call('ZADD', key, now, now .. '-' .. math.random())
        redis.call('EXPIRE', key, window)
        return {1, limit - current - 1, window}
    else
        -- Get time until oldest entry expires
        local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
        local retry_after = oldest[2] + window - now
        return {0, 0, retry_after}
    end
    """
    
    # Lua script for token bucket
    TOKEN_BUCKET_SCRIPT = """
    local key = KEYS[1]
    local now = tonumber(ARGV[1])
    local rate = tonumber(ARGV[2])
    local capacity = tonumber(ARGV[3])
    
    local bucket = redis.call('HMGET', key, 'tokens', 'last_update')
    local tokens = tonumber(bucket[1]) or capacity
    local last_update = tonumber(bucket[2]) or now
    
    -- Calculate tokens to add
    local elapsed = now - last_update
    tokens = math.min(capacity, tokens + elapsed * rate)
    
    if tokens >= 1 then
        tokens = tokens - 1
        redis.call('HMSET', key, 'tokens', tokens, 'last_update', now)
        redis.call('EXPIRE', key, math.ceil(capacity / rate) * 2)
        return {1, math.floor(tokens), 0}
    else
        local retry_after = (1 - tokens) / rate
        return {0, 0, retry_after}
    end
    """
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self._sliding_window_sha = None
        self._token_bucket_sha = None
    
    async def init_scripts(self):
        """Load Lua scripts into Redis"""
        self._sliding_window_sha = await self.redis.script_load(
            self.SLIDING_WINDOW_SCRIPT
        )
        self._token_bucket_sha = await self.redis.script_load(
            self.TOKEN_BUCKET_SCRIPT
        )
    
    async def is_allowed(
        self,
        identifier: str,
        config: RateLimitConfig
    ) -> tuple[bool, dict]:
        """Check if request is allowed under rate limit"""
        key = f"{config.key_prefix}:{identifier}"
        now = time.time()
        
        if config.algorithm == RateLimitAlgorithm.SLIDING_WINDOW:
            result = await self.redis.evalsha(
                self._sliding_window_sha,
                1, key,
                now, config.window_seconds, config.requests
            )
        elif config.algorithm == RateLimitAlgorithm.TOKEN_BUCKET:
            rate = config.requests / config.window_seconds
            result = await self.redis.evalsha(
                self._token_bucket_sha,
                1, key,
                now, rate, config.requests
            )
        else:  # Fixed window
            result = await self._fixed_window(key, config)
        
        allowed, remaining, retry_after = result
        
        return bool(allowed), {
            "limit": config.requests,
            "remaining": int(remaining),
            "reset": int(now + config.window_seconds),
            "retry_after": int(retry_after) if not allowed else None
        }
    
    async def _fixed_window(
        self,
        key: str,
        config: RateLimitConfig
    ) -> tuple:
        """Simple fixed window rate limiting"""
        current = await self.redis.incr(key)
        if current == 1:
            await self.redis.expire(key, config.window_seconds)
        
        if current <= config.requests:
            return (1, config.requests - current, 0)
        else:
            ttl = await self.redis.ttl(key)
            return (0, 0, ttl)


# FastAPI dependency
def rate_limit(
    requests: int = 100,
    window_seconds: int = 60,
    key_func: Optional[Callable[[Request], str]] = None
):
    """Rate limiting dependency with customizable key extraction"""
    
    config = RateLimitConfig(
        requests=requests,
        window_seconds=window_seconds
    )
    
    async def dependency(request: Request):
        rate_limiter: RateLimiter = request.app.state.rate_limiter
        
        # Extract identifier (IP, user ID, API key, etc.)
        if key_func:
            identifier = key_func(request)
        else:
            # Default: use IP + endpoint
            ip = request.client.host
            path = request.url.path
            identifier = hashlib.md5(f"{ip}:{path}".encode()).hexdigest()
        
        allowed, info = await rate_limiter.is_allowed(identifier, config)
        
        # Add rate limit headers to response
        request.state.rate_limit_headers = {
            "X-RateLimit-Limit": str(info["limit"]),
            "X-RateLimit-Remaining": str(info["remaining"]),
            "X-RateLimit-Reset": str(info["reset"])
        }
        
        if not allowed:
            raise HTTPException(
                status_code=429,
                detail={
                    "code": "RATE_LIMIT_EXCEEDED",
                    "message": f"Rate limit exceeded. Retry after {info['retry_after']} seconds.",
                    "retry_after": info["retry_after"]
                },
                headers={
                    "Retry-After": str(info["retry_after"]),
                    **request.state.rate_limit_headers
                }
            )
        
        return info
    
    return Depends(dependency)


# Usage
app = FastAPI()

@app.on_event("startup")
async def startup():
    app.state.redis = redis.from_url("redis://localhost:6379")
    app.state.rate_limiter = RateLimiter(app.state.redis)
    await app.state.rate_limiter.init_scripts()

# Global rate limit
@app.get("/api/data", dependencies=[rate_limit(requests=100, window_seconds=60)])
async def get_data():
    return {"data": "value"}

# Custom key (per-user rate limit)
def user_key(request: Request) -> str:
    return f"user:{request.state.user_id}"

@app.get("/api/premium", dependencies=[rate_limit(requests=1000, window_seconds=60, key_func=user_key)])
async def premium_endpoint():
    return {"data": "premium"}

# Tiered rate limits
@app.get("/api/search")
async def search(
    q: str,
    request: Request,
    _free: dict = rate_limit(requests=10, window_seconds=60),
    _burst: dict = rate_limit(requests=5, window_seconds=1)  # Also limit bursts
):
    return {"results": []}