from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
import uvicorn

app = FastAPI(
    title="User API",
    description="RESTful API 示例",
    version="1.0.0"
)

# 数据模型
class UserBase(BaseModel):
    username: str = Field(..., min_length=1, max_length=50)
    email: str = Field(..., regex=r"^[\w\.-]+@[\w\.-]+\.\w+$")
    age: Optional[int] = Field(None, ge=0, le=150)

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

class UserUpdate(BaseModel):
    username: Optional[str] = None
    email: Optional[str] = None
    age: Optional[int] = None

class User(UserBase):
    id: int
    created_at: datetime

    class Config:
        json_schema_extra = {
            "example": {
                "id": 1,
                "username": "john_doe",
                "email": "john@example.com",
                "age": 30,
                "created_at": "2025-01-01T12:00:00"
            }
        }

# 内存数据库（实际应使用真实数据库）
fake_db = {}
user_id_counter = 1

# API 端点

@app.get("/api/v1/users", response_model=List[User], tags=["Users"])
async def list_users(skip: int = 0, limit: int = 10):
    """
    列出所有用户
    
    - **skip**: 跳过的记录数
    - **limit**: 返回的最大记录数
    """
    users = list(fake_db.values())
    return users[skip:skip + limit]

@app.get("/api/v1/users/{user_id}", response_model=User, tags=["Users"])
async def get_user(user_id: int):
    """获取单个用户信息"""
    if user_id not in fake_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"用户 ID {user_id} 不存在"
        )
    return fake_db[user_id]

@app.post("/api/v1/users", response_model=User, status_code=status.HTTP_201_CREATED, tags=["Users"])
async def create_user(user: UserCreate):
    """创建新用户"""
    global user_id_counter
    
    # 检查邮箱重复
    for existing_user in fake_db.values():
        if existing_user["email"] == user.email:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail=f"邮箱 {user.email} 已被使用"
            )
    
    new_user = {
        "id": user_id_counter,
        "username": user.username,
        "email": user.email,
        "age": user.age,
        "created_at": datetime.now()
    }
    
    fake_db[user_id_counter] = new_user
    user_id_counter += 1
    
    return new_user

@app.put("/api/v1/users/{user_id}", response_model=User, tags=["Users"])
async def update_user(user_id: int, user_update: UserUpdate):
    """完全更新用户信息"""
    if user_id not in fake_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"用户 ID {user_id} 不存在"
        )
    
    existing_user = fake_db[user_id]
    update_data = user_update.dict(exclude_unset=True)
    updated_user = {**existing_user, **update_data}
    
    fake_db[user_id] = updated_user
    return updated_user

@app.patch("/api/v1/users/{user_id}", response_model=User, tags=["Users"])
async def partial_update_user(user_id: int, user_update: UserUpdate):
    """部分更新用户信息"""
    # 与 PUT 相同，因为我们使用 exclude_unset=True
    return await update_user(user_id, user_update)

@app.delete("/api/v1/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["Users"])
async def delete_user(user_id: int):
    """删除用户"""
    if user_id not in fake_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"用户 ID {user_id} 不存在"
        )
    
    del fake_db[user_id]

# 错误处理
@app.exception_handler(ValueError)
async def value_error_exception_handler(request, exc):
    return HTTPException(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        detail=str(exc)
    )

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

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

app = FastAPI()

@app.get("/api/v1/static-data")
async def get_static_data(response: Response):
    """获取静态数据，设置长期缓存"""
    response.headers["Cache-Control"] = "public, max-age=31536000"  # 1 年
    response.headers["ETag"] = '"123456"'
    return {"data": "static content"}

@app.get("/api/v1/dynamic-data")
async def get_dynamic_data(response: Response):
    """获取动态数据，设置短期缓存"""
    response.headers["Cache-Control"] = "public, max-age=300"  # 5 分钟
    return {"data": "dynamic content", "timestamp": datetime.now()}

@app.get("/api/v1/no-cache-data")
async def get_no_cache_data(response: Response):
    """获取不可缓存的数据"""
    response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
    response.headers["Pragma"] = "no-cache"
    response.headers["Expires"] = "0"
    return {"data": "sensitive content"}

import strawberry
from typing import List, Optional
from datetime import datetime
from dataclasses import dataclass

# 数据模型
@dataclass
class UserModel:
    id: int
    username: str
    email: str
    age: Optional[int] = None
    created_at: datetime = None

# Strawberry 类型定义
@strawberry.type
class User:
    id: strawberry.ID
    username: str
    email: str
    age: Optional[int] = None
    created_at: datetime = None

    @strawberry.field
    def posts(self) -> List["Post"]:
        """获取用户的文章列表"""
        # 这里会调用 dataloader 或数据库查询
        return []

@strawberry.type
class Post:
    id: strawberry.ID
    title: str
    content: str
    author_id: int
    created_at: datetime = None

    @strawberry.field
    def author(self) -> User:
        """获取文章作者"""
        # 这里会调用 dataloader 避免 N+1 查询
        return User(
            id=strawberry.ID(str(self.author_id)),
            username="author_name",
            email="author@example.com"
        )

# 查询类型
@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: strawberry.ID) -> Optional[User]:
        """获取单个用户"""
        # 从数据库查询
        return User(
            id=id,
            username="john_doe",
            email="john@example.com",
            age=30,
            created_at=datetime.now()
        )

    @strawberry.field
    def users(
        self,
        first: int = 10,
        skip: int = 0
    ) -> List[User]:
        """获取用户列表"""
        # 从数据库查询
        return [
            User(
                id=strawberry.ID(str(i)),
                username=f"user_{i}",
                email=f"user_{i}@example.com"
            )
            for i in range(skip, skip + first)
        ]

# 变更类型
@strawberry.input
class CreateUserInput:
    username: str
    email: str
    password: str
    age: Optional[int] = None

@strawberry.type
class CreateUserPayload:
    user: Optional[User]
    success: bool
    message: str

@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_user(self, input: CreateUserInput) -> CreateUserPayload:
        """创建用户"""
        # 验证输入
        if not input.email or "@" not in input.email:
            return CreateUserPayload(
                user=None,
                success=False,
                message="邮箱格式错误"
            )
        
        # 创建用户（实际应保存到数据库）
        new_user = User(
            id=strawberry.ID("1"),
            username=input.username,
            email=input.email,
            age=input.age
        )
        
        return CreateUserPayload(
            user=new_user,
            success=True,
            message="用户创建成功"
        )

# 创建 Schema
schema = strawberry.Schema(query=Query, mutation=Mutation)

# 使用 FastAPI 启动 GraphQL 服务
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

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

import asyncio
from strawberry import execute

query = """
    query {
        user(id: "1") {
            id
            username
            email
            posts {
                id
                title
            }
        }
    }
"""

result = asyncio.run(schema.execute(query))
print(result.data)

# ❌ 不好的实现：每个用户都会查询一次文章（N+1 问题）
@strawberry.type
class User:
    @strawberry.field
    def posts(self) -> List[Post]:
        # 这会为每个用户执行一次查询
        return db.query(Post).filter(Post.author_id == self.id).all()

from strawberry.dataloader import DataLoader
from typing import List

# 创建 DataLoader
async def load_posts_for_users(user_ids: List[int]) -> List[List[Post]]:
    """批量加载用户的文章"""
    posts_by_user = db.query(Post).filter(Post.author_id.in_(user_ids)).all()
    
    # 按用户 ID 分组
    result = {user_id: [] for user_id in user_ids}
    for post in posts_by_user:
        result[post.author_id].append(post)
    
    return [result[user_id] for user_id in user_ids]

posts_loader = DataLoader(load_posts_for_users)

# 使用 DataLoader
@strawberry.type
class User:
    @strawberry.field
    async def posts(self, loader: DataLoader) -> List[Post]:
        # 通过 DataLoader 批量加载，只执行 1 次查询
        return await posts_loader.load(int(self.id))

import grpc
from concurrent import futures
import logging
from datetime import datetime

# 假设已通过 protoc 生成了 pb2 和 pb2_grpc 文件
import user_pb2
import user_pb2_grpc

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

class UserServicer(user_pb2_grpc.UserServiceServicer):
    """用户服务实现"""
    
    def __init__(self):
        # 模拟数据库
        self.users = {}
        self.user_id_counter = 1
    
    def GetUser(self, request, context):
        """获取单个用户"""
        logger.info(f"Getting user {request.id}")
        
        if request.id not in self.users:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f"User {request.id} not found")
            return user_pb2.GetUserResponse()
        
        return user_pb2.GetUserResponse(
            user=self.users[request.id]
        )
    
    def ListUsers(self, request, context):
        """服务器流：列出用户"""
        logger.info(f"Listing users: skip={request.skip}, limit={request.limit}")
        
        users = list(self.users.values())
        for user in users[request.skip:request.skip + request.limit]:
            yield user
    
    def CreateUsers(self, request_iterator, context):
        """客户端流：批量创建用户"""
        created_users = []
        
        try:
            for request in request_iterator:
                logger.info(f"Creating user: {request.username}")
                
                user = user_pb2.User(
                    id=self.user_id_counter,
                    username=request.username,
                    email=request.email,
                    age=request.age,
                    created_at=datetime.now().isoformat()
                )
                
                self.users[self.user_id_counter] = user
                created_users.append(user)
                self.user_id_counter += 1
        
        except Exception as e:
            context.set_code(grpc.StatusCode.INTERNAL)
            context.set_details(str(e))
        
        return user_pb2.CreateUserResponse(
            success=len(created_users) > 0,
            message=f"Created {len(created_users)} users"
        )
    
    def UpdateUsers(self, request_iterator, context):
        """双向流：更新用户"""
        for updated_user in request_iterator:
            if updated_user.id in self.users:
                logger.info(f"Updating user {updated_user.id}")
                self.users[updated_user.id] = updated_user
                yield updated_user
            else:
                context.set_code(grpc.StatusCode.NOT_FOUND)
                context.set_details(f"User {updated_user.id} not found")
    
    def CreateUser(self, request, context):
        """创建单个用户"""
        logger.info(f"Creating user: {request.username}")
        
        # 验证邮箱唯一性
        for user in self.users.values():
            if user.email == request.email:
                context.set_code(grpc.StatusCode.ALREADY_EXISTS)
                context.set_details(f"Email {request.email} already exists")
                return user_pb2.CreateUserResponse()
        
        user = user_pb2.User(
            id=self.user_id_counter,
            username=request.username,
            email=request.email,
            age=request.age,
            created_at=datetime.now().isoformat()
        )
        
        self.users[self.user_id_counter] = user
        self.user_id_counter += 1
        
        return user_pb2.CreateUserResponse(
            user=user,
            success=True,
            message="User created successfully"
        )
    
    def DeleteUser(self, request, context):
        """删除用户"""
        logger.info(f"Deleting user {request.id}")
        
        if request.id in self.users:
            del self.users[request.id]
            return user_pb2.DeleteUserResponse(success=True)
        else:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f"User {request.id} not found")
            return user_pb2.DeleteUserResponse(success=False)

def serve():
    """启动 gRPC 服务器"""
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    user_pb2_grpc.add_UserServiceServicer_to_server(
        UserServicer(), server
    )
    
    server.add_insecure_port("[::]:50051")
    logger.info("gRPC server started on port 50051")
    server.start()
    server.wait_for_termination()

if __name__ == "__main__":
    serve()

import grpc
import user_pb2
import user_pb2_grpc
import logging

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

def run():
    """gRPC 客户端示例"""
    
    # 创建连接
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = user_pb2_grpc.UserServiceStub(channel)
        
        # 1. 一元 RPC：创建用户
        logger.info("Creating users...")
        for i in range(3):
            request = user_pb2.CreateUserRequest(
                username=f"user_{i}",
                email=f"user_{i}@example.com",
                age=20 + i
            )
            response = stub.CreateUser(request)
            logger.info(f"Created: {response.user.username}")
        
        # 2. 一元 RPC：获取单个用户
        logger.info("Getting user 1...")
        request = user_pb2.GetUserRequest(id=1)
        response = stub.GetUser(request)
        logger.info(f"Got: {response.user.username}, {response.user.email}")
        
        # 3. 服务器流 RPC：列出用户
        logger.info("Listing users...")
        request = user_pb2.ListUsersRequest(skip=0, limit=10)
        for user in stub.ListUsers(request):
            logger.info(f"User: {user.username}")
        
        # 4. 客户端流 RPC：批量创建用户
        logger.info("Creating users in batch...")
        def request_generator():
            for i in range(3, 6):
                yield user_pb2.CreateUserRequest(
                    username=f"batch_user_{i}",
                    email=f"batch_user_{i}@example.com",
                    age=30 + i
                )
        
        response = stub.CreateUsers(request_generator())
        logger.info(f"Batch result: {response.message}")
        
        # 5. 双向流 RPC：更新用户
        logger.info("Updating users...")
        def update_generator():
            for i in range(1, 4):
                yield user_pb2.User(
                    id=i,
                    username=f"updated_user_{i}",
                    email=f"updated_{i}@example.com",
                    age=25 + i,
                    created_at="2025-01-01T00:00:00"
                )
        
        for updated_user in stub.UpdateUsers(update_generator()):
            logger.info(f"Updated: {updated_user.username}")

if __name__ == "__main__":
    run()

import time
import requests
import statistics
from typing import List

def benchmark_rest():
    """RESTful API 性能测试"""
    times = []
    
    for _ in range(100):
        start = time.time()
        # 获取用户列表
        requests.get("http://localhost:8000/api/v1/users?limit=100")
        # 对每个用户获取文章（N+1 问题）
        for user_id in range(1, 101):
            requests.get(f"http://localhost:8000/api/v1/users/{user_id}/posts")
        times.append(time.time() - start)
    
    return {
        "avg": statistics.mean(times),
        "median": statistics.median(times),
        "stdev": statistics.stdev(times),
        "min": min(times),
        "max": max(times)
    }

def benchmark_graphql():
    """GraphQL 性能测试"""
    times = []
    query = """
        query {
            users(first: 100) {
                edges {
                    node {
                        id
                        username
                        posts(first: 10) {
                            id
                            title
                        }
                    }
                }
            }
        }
    """
    
    for _ in range(100):
        start = time.time()
        requests.post(
            "http://localhost:4000/graphql",
            json={"query": query}
        )
        times.append(time.time() - start)
    
    return {
        "avg": statistics.mean(times),
        "median": statistics.median(times),
        "stdev": statistics.stdev(times),
        "min": min(times),
        "max": max(times)
    }

def benchmark_grpc():
    """gRPC 性能测试"""
    times = []
    
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = user_pb2_grpc.UserServiceStub(channel)
        
        for _ in range(100):
            start = time.time()
            request = user_pb2.ListUsersRequest(skip=0, limit=100)
            for user in stub.ListUsers(request):
                pass
            times.append(time.time() - start)
    
    return {
        "avg": statistics.mean(times),
        "median": statistics.median(times),
        "stdev": statistics.stdev(times),
        "min": min(times),
        "max": max(times)
    }

# 运行基准测试
if __name__ == "__main__":
    print("RESTful:", benchmark_rest())
    print("GraphQL:", benchmark_graphql())
    print("gRPC:", benchmark_grpc())

# RESTful - JWT 认证
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthenticationCredentials
import jwt

security = HTTPBearer()

async def verify_token(credentials: HTTPAuthenticationCredentials = Depends(security)):
    try:
        payload = jwt.decode(credentials.credentials, "secret_key", algorithms=["HS256"])
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401)
        return user_id
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401)

@app.get("/api/v1/users/me")
async def get_current_user(user_id: str = Depends(verify_token)):
    return {"user_id": user_id}

# gRPC - 拦截器认证
class AuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        # 从元数据中提取令牌
        metadata = dict(handler_call_details.invocation_metadata)
        token = metadata.get("authorization")
        
        if not token or not verify_token(token):
            raise grpc.RpcError(
                grpc.StatusCode.UNAUTHENTICATED,
                "Invalid token"
            )
        
        return continuation(handler_call_details)

server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
server.add_generic_rpc_interceptor(AuthInterceptor())

# RESTful - 使用 Pydantic
from pydantic import BaseModel, EmailStr, Field

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=8)
    age: int = Field(..., ge=18, le=150)

# GraphQL - 使用 validation 装饰器
@strawberry.input
class CreateUserInput:
    username: str  # Strawberry 会自动验证类型
    email: str
    password: str
    
    def __post_init__(self):
        if len(self.username) < 3:
            raise ValueError("Username must be at least 3 characters")

# gRPC - 在服务实现中验证
def validate_user(user: user_pb2.User) -> bool:
    if len(user.username) < 3:
        return False
    if "@" not in user.email:
        return False
    if user.age < 18 or user.age > 150:
        return False
    return True

# RESTful - 使用 OpenTelemetry
from opentelemetry import trace, metrics
from opentelemetry.exporter.jaeger import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

jaeger_exporter = JaegerExporter(agent_host_name="localhost")
trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

FastAPIInstrumentor.instrument_app(app)

# 统一的日志格式
import structlog

structlog.configure(
    processors=[
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)

logger = structlog.get_logger()

# 在 API 中记录日志
@app.post("/api/v1/users")
async def create_user(user: UserCreate):
    logger.info("user.create", username=user.username, email=user.email)
    # ...