flowchart TB
    subgraph "外部层 External Layer"
        Client[客户端 Client]
        ASGIServer[ASGI 服务器<br/>Uvicorn/Hypercorn]
    end
    
    subgraph "应用层 Application Layer"
        FastAPIApp[FastAPI 应用实例<br/>applications.py]
        Router[路由系统<br/>routing.py]
        Middleware[中间件栈<br/>middleware/]
    end
    
    subgraph "核心处理层 Core Processing Layer"
        DependencyInjection[依赖注入系统<br/>dependencies/]
        Validation[请求验证<br/>Pydantic Models]
        ParamExtraction[参数提取<br/>params.py]
        ResponseHandling[响应处理<br/>responses.py]
    end
    
    subgraph "文档生成层 Documentation Layer"
        OpenAPIGen[OpenAPI 生成<br/>openapi/]
        DocsUI[文档 UI<br/>Swagger/ReDoc]
    end
    
    subgraph "安全层 Security Layer"
        SecuritySchemes[安全方案<br/>security/]
        OAuth2[OAuth2/JWT]
        APIKey[API Key]
    end
    
    subgraph "底层基础 Infrastructure"
        Starlette[Starlette 框架]
        Pydantic[Pydantic 验证]
        Encoders[JSON 编码器<br/>encoders.py]
    end
    
    Client --> ASGIServer
    ASGIServer --> FastAPIApp
    FastAPIApp --> Middleware
    Middleware --> Router
    Router --> DependencyInjection
    DependencyInjection --> ParamExtraction
    DependencyInjection --> SecuritySchemes
    ParamExtraction --> Validation
    Validation --> ResponseHandling
    ResponseHandling --> Encoders
    
    FastAPIApp -.生成.-> OpenAPIGen
    OpenAPIGen -.提供.-> DocsUI
    
    SecuritySchemes --> OAuth2
    SecuritySchemes --> APIKey
    
    Router -.依赖.-> Starlette
    Validation -.依赖.-> Pydantic
    FastAPIApp -.继承.-> Starlette

HTTP 请求 → ASGI Scope → Starlette Request → 参数提取 → Pydantic 验证 → 业务函数参数

业务函数返回值 → Response Model 验证 → JSON 编码 → Starlette Response → ASGI Send → HTTP 响应

sequenceDiagram
    autonumber
    participant C as 客户端 Client
    participant AS as ASGI 服务器<br/>Uvicorn
    participant FA as FastAPI 应用
    participant MW as 中间件栈
    participant R as 路由系统
    participant DI as 依赖注入
    participant V as Pydantic 验证
    participant H as 路由处理函数
    participant RM as 响应模型验证
    participant E as JSON 编码器
    
    C->>AS: HTTP 请求<br/>POST /items
    AS->>FA: ASGI 调用<br/>(scope, receive, send)
    FA->>MW: 中间件入口
    Note over MW: CORS、异常处理<br/>等中间件
    MW->>R: 路由匹配
    R->>R: 根据路径和方法<br/>匹配路由
    R->>DI: 解析依赖树
    
    loop 遍历依赖项
        DI->>DI: 检查缓存
        alt 已缓存
            DI->>DI: 使用缓存值
        else 未缓存
            DI->>V: 提取参数
            V->>V: 类型验证与转换
            V-->>DI: 验证后的参数
            DI->>DI: 缓存依赖结果
        end
    end
    
    DI->>V: 提取路由参数
    Note over V: 路径参数、查询参数<br/>请求体、头部、Cookie
    V->>V: Pydantic 模型验证
    
    alt 验证失败
        V-->>C: 422 验证错误
    else 验证成功
        V-->>DI: 已验证参数
        DI->>H: 调用处理函数<br/>注入依赖和参数
        
        alt 异步函数
            H->>H: await 执行
        else 同步函数
            H->>H: 线程池执行
        end
        
        H-->>RM: 返回值
        
        opt 声明了响应模型
            RM->>RM: 验证返回值
            alt 验证失败
                RM-->>C: 500 内部错误
            end
        end
        
        RM->>E: 序列化响应
        E->>E: 转换特殊类型<br/>(datetime, UUID, Enum)
        E-->>MW: JSON 响应
        MW-->>AS: HTTP 响应
        AS-->>C: 200 OK + JSON
    end

@app.post("/items/")
async def create_item(
    item: Item,
    idempotency_key: str = Header(...),
    db: Session = Depends(get_db)
):
    # 检查幂等键是否已存在
    existing = db.query(IdempotencyRecord).filter_by(key=idempotency_key).first()
    if existing:
        return existing.response  # 返回缓存的响应
    
    # 执行业务逻辑
    result = create_item_in_db(item, db)
    
    # 存储幂等记录
    db.add(IdempotencyRecord(key=idempotency_key, response=result))
    db.commit()
    
    return result

import asyncio

async def timeout_middleware(request: Request, call_next):
    try:
        return await asyncio.wait_for(call_next(request), timeout=30.0)
    except asyncio.TimeoutError:
        return JSONResponse({"detail": "Request timeout"}, status_code=504)

app.middleware("http")(timeout_middleware)

async def get_external_data():
    async with httpx.AsyncClient(timeout=5.0) as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

flowchart TB
    subgraph "01-应用与路由模块"
        M01[applications.py<br/>FastAPI 主应用类]
        M02[routing.py<br/>APIRouter/APIRoute]
    end
    
    subgraph "02-依赖注入模块"
        M03[dependencies/utils.py<br/>依赖解析]
        M04[dependencies/models.py<br/>Dependant 模型]
    end
    
    subgraph "03-参数处理模块"
        M05[params.py<br/>参数类定义]
        M06[param_functions.py<br/>参数装饰器函数]
    end
    
    subgraph "04-请求响应模块"
        M07[requests.py<br/>Request 封装]
        M08[responses.py<br/>Response 类型]
        M09[encoders.py<br/>JSON 编码器]
    end
    
    subgraph "05-验证模块"
        M10[Pydantic 模型<br/>外部依赖]
    end
    
    subgraph "06-OpenAPI 模块"
        M11[openapi/utils.py<br/>OpenAPI 生成]
        M12[openapi/docs.py<br/>文档 UI]
        M13[openapi/models.py<br/>OpenAPI 模型]
    end
    
    subgraph "07-安全模块"
        M14[security/oauth2.py<br/>OAuth2 方案]
        M15[security/api_key.py<br/>API Key]
        M16[security/http.py<br/>HTTP 认证]
    end
    
    subgraph "08-中间件模块"
        M17[middleware/cors.py<br/>CORS]
        M18[middleware/gzip.py<br/>GZIP]
        M19[middleware/trustedhost.py<br/>可信主机]
    end
    
    subgraph "09-异常处理模块"
        M20[exceptions.py<br/>异常类]
        M21[exception_handlers.py<br/>异常处理器]
    end
    
    subgraph "10-辅助模块"
        M22[background.py<br/>后台任务]
        M23[datastructures.py<br/>数据结构]
        M24[websockets.py<br/>WebSocket]
        M25[testclient.py<br/>测试客户端]
    end
    
    M01 -->|注册路由| M02
    M01 -->|添加中间件| M17
    M01 -->|生成 OpenAPI| M11
    M01 -->|注册异常处理器| M21
    
    M02 -->|解析依赖| M03
    M02 -->|提取参数| M05
    M02 -->|处理请求| M07
    M02 -->|返回响应| M08
    
    M03 -->|构建 Dependant| M04
    M03 -->|验证数据| M10
    M03 -->|使用参数类| M05
    
    M06 -->|创建参数实例| M05
    
    M08 -->|序列化| M09
    
    M11 -->|分析路由| M02
    M11 -->|分析参数| M05
    M11 -->|分析模型| M10
    M11 -->|生成模型| M13
    M12 -->|使用 OpenAPI| M11
    
    M14 -->|作为依赖| M03
    M15 -->|作为依赖| M03
    M16 -->|作为依赖| M03
    
    M02 -->|支持后台任务| M22
    M02 -->|处理文件上传| M23

app.state.db_pool = create_pool()  # 数据库连接池
app.state.redis = redis.Redis()    # Redis 客户端

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request.state.request_id = generate_id()
    response = await call_next(request)
    return response

async def get_db():
    async with SessionLocal() as session:
        yield session  # 请求结束时自动关闭

@app.get("/items/")
async def read_items(db: Session = Depends(get_db)):
    return db.query(Item).all()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    # 订阅消息

async def get_db():
    async with SessionLocal() as session:
        async with session.begin():  # 事务边界
            yield session

async def get_db_with_transaction():
    async with SessionLocal() as session:
        async with session.begin():
            yield session
            # 自动提交或回滚

@app.post("/items/")
async def create_item(item: Item, db: Session = Depends(get_db)):
    async with db.begin():
        db.add(item)
        # 手动管理事务

class Item(Base):
    id: int
    version: int  # 版本号
    
@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: ItemUpdate,
    db: Session = Depends(get_db)
):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if db_item.version != item.version:
        raise HTTPException(409, "Conflict: item was modified")
    db_item.version += 1
    # 更新其他字段

# PostgreSQL 行锁
db_item = db.query(Item).filter(Item.id == item_id).with_for_update().first()

async def get_lock():
    redis = redis.Redis()
    lock = redis.lock("resource:123", timeout=10)
    acquired = lock.acquire(blocking=False)
    if not acquired:
        raise HTTPException(423, "Resource locked")
    try:
        yield lock
    finally:
        lock.release()

@app.get("/dashboard")
async def dashboard():
    users, items, orders = await asyncio.gather(
        get_users(),
        get_items(),
        get_orders()
    )
    return {"users": users, "items": items, "orders": orders}

# 避免阻塞
async def get_user(user_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        return response.json()

# 依赖项默认在请求中缓存
async def get_db():
    yield SessionLocal()  # 同一请求中只创建一次

# 排除未设置字段，减少序列化开销
@app.get("/items/", response_model=List[Item], response_model_exclude_unset=True)
async def read_items():
    return items

@app.get("/large-file")
async def stream_large_file():
    async def generate():
        for chunk in read_large_file_in_chunks():
            yield chunk
    return StreamingResponse(generate())

from prometheus_client import Counter, Histogram, generate_latest

request_count = Counter("http_requests_total", "Total requests", ["method", "endpoint"])
request_duration = Histogram("http_request_duration_seconds", "Request duration")

@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    with request_duration.time():
        response = await call_next(request)
    request_count.labels(method=request.method, endpoint=request.url.path).inc()
    return response

@app.get("/metrics")
async def metrics():
    return Response(generate_latest(), media_type="text/plain")

from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

FastAPIInstrumentor.instrument_app(app)

app = FastAPI(
    title="My API",                    # API 标题
    version="1.0.0",                   # API 版本
    openapi_url="/api/v1/openapi.json", # OpenAPI JSON 路径
    docs_url="/docs",                  # Swagger UI 路径
    redoc_url="/redoc",                # ReDoc 路径
    debug=False,                       # 调试模式
)

@app.get(
    "/items/",
    response_model=List[Item],         # 响应模型
    status_code=200,                   # 默认状态码
    tags=["items"],                    # OpenAPI 标签
    summary="Get all items",           # 摘要
    description="Retrieve all items",  # 详细描述
    deprecated=False,                  # 是否废弃
    include_in_schema=True,            # 是否包含在 OpenAPI
)
async def read_items():
    pass

async def pagination(skip: int = Query(0, ge=0), limit: int = Query(10, le=100)):
    return {"skip": skip, "limit": limit}

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    redis_url: str
    secret_key: str
    
    class Config:
        env_file = ".env"

settings = Settings()

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

uvicorn main:app --reload

curl http://localhost:8000/
curl http://localhost:8000/items/42?q=test

from typing import List, Optional
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel

app = FastAPI()

# 数据模型
class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    
    class Config:
        from_attributes = True

# 模拟数据库
fake_db = {}
next_id = 1

# 依赖项
def get_item_or_404(item_id: int) -> Item:
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]

# CRUD 操作
@app.post("/items/", response_model=Item, status_code=201)
async def create_item(item: ItemCreate):
    global next_id
    db_item = Item(id=next_id, **item.model_dump())
    fake_db[next_id] = db_item
    next_id += 1
    return db_item

@app.get("/items/", response_model=List[Item])
async def list_items(skip: int = 0, limit: int = 10):
    return list(fake_db.values())[skip:skip+limit]

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item: Item = Depends(get_item_or_404)):
    return item

@app.put("/items/{item_id}", response_model=Item)
async def update_item(
    item_id: int,
    item_update: ItemCreate,
    existing_item: Item = Depends(get_item_or_404)
):
    updated_item = Item(id=item_id, **item_update.model_dump())
    fake_db[item_id] = updated_item
    return updated_item

@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int, item: Item = Depends(get_item_or_404)):
    del fake_db[item_id]

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from fastapi import FastAPI, Depends

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 模型
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    name = Column(String)

Base.metadata.create_all(bind=engine)

# Pydantic 模型
from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    name: str

class UserOut(BaseModel):
    id: int
    email: str
    name: str
    
    class Config:
        from_attributes = True

app = FastAPI()

# 依赖项：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 路由
@app.post("/users/", response_model=UserOut)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(**user.model_dump())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/users/{user_id}", response_model=UserOut)
async def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

from datetime import datetime, timedelta
from typing import Optional
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

# 模型
class Token(BaseModel):
    access_token: str
    token_type: str

class User(BaseModel):
    username: str
    email: Optional[str] = None
    disabled: Optional[bool] = False

fake_users_db = {
    "testuser": {
        "username": "testuser",
        "email": "test@example.com",
        "hashed_password": pwd_context.hash("secret"),
        "disabled": False,
    }
}

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = fake_users_db.get(username)
    if user is None:
        raise credentials_exception
    return User(**user)

# 登录端点
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(form_data.username)
    if not user or not verify_password(form_data.password, user["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token = create_access_token(
        data={"sub": user["username"]},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

# 受保护端点
@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

@app.get("/items/")
async def read_items(current_user: User = Depends(get_current_user)):
    return [{"item": "Item 1", "owner": current_user.username}]

# 1. 获取 token
curl -X POST "http://localhost:8000/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "username=testuser&password=secret"

# 返回: {"access_token": "eyJ...", "token_type": "bearer"}

# 2. 使用 token 访问受保护资源
curl -X GET "http://localhost:8000/users/me" \
  -H "Authorization: Bearer eyJ..."

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, EmailStr

app = FastAPI()

class EmailSchema(BaseModel):
    email: EmailStr
    message: str

def send_email(email: str, message: str):
    # 模拟耗时的邮件发送
    import time
    time.sleep(3)
    print(f"Email sent to {email}: {message}")

@app.post("/send-notification/")
async def send_notification(
    email: EmailSchema,
    background_tasks: BackgroundTasks
):
    # 添加后台任务
    background_tasks.add_task(send_email, email.email, email.message)
    
    # 立即返回响应（不等待邮件发送完成）
    return {"message": "Notification sent in the background"}

@app.post("/send-multiple-notifications/")
async def send_multiple_notifications(
    emails: list[EmailSchema],
    background_tasks: BackgroundTasks
):
    for email in emails:
        background_tasks.add_task(send_email, email.email, email.message)
    
    return {"message": f"{len(emails)} notifications scheduled"}

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

@app.middleware("http")
async def log_requests(request: Request, call_next):
    print(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    print(f"Response: {response.status_code}")
    return response

@app.get("/")
async def root():
    return {"message": "Hello World"}

请求 → log_requests (before) → add_process_time_header (before) 
     → 路由处理 
     → add_process_time_header (after) → log_requests (after) → 响应

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client #{client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left")

const ws = new WebSocket("ws://localhost:8000/ws/123");

ws.onmessage = (event) => {
    console.log("Received:", event.data);
};

ws.send("Hello Server!");

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import FileResponse
import shutil
from pathlib import Path

app = FastAPI()

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    file_path = UPLOAD_DIR / file.filename
    with file_path.open("wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": file_path.stat().st_size
    }

@app.post("/upload-multiple/")
async def upload_multiple_files(files: list[UploadFile] = File(...)):
    uploaded_files = []
    for file in files:
        file_path = UPLOAD_DIR / file.filename
        with file_path.open("wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
        uploaded_files.append(file.filename)
    
    return {"uploaded_files": uploaded_files}

@app.get("/download/{filename}")
async def download_file(filename: str):
    file_path = UPLOAD_DIR / filename
    if not file_path.exists():
        raise HTTPException(status_code=404, detail="File not found")
    
    return FileResponse(
        path=file_path,
        filename=filename,
        media_type="application/octet-stream"
    )

from typing import Optional
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    is_offer: Optional[bool] = None

@app.post("/items/{item_id}")
async def create_item(
    item_id: int,              # 路径参数（从 URL 提取）
    item: Item,                # 请求体（从 JSON 解析）
    q: Optional[str] = None,   # 查询参数（从 query string 提取）
):
    # FastAPI 自动：
    # 1. 从 URL 提取 item_id 并转换为 int
    # 2. 解析 JSON 请求体为 Item 模型
    # 3. 验证 Item 字段（name 必须是 str，price 必须是 float）
    # 4. 从 query string 提取可选参数 q
    # 5. 生成 OpenAPI 文档
    return {"item_id": item_id, "item": item, "q": q}

# 带清理的依赖项（使用 yield）
async def get_db():
    db = Database()
    try:
        yield db
    finally:
        await db.close()

# 类作为依赖项
class Pagination:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip = skip
        self.limit = limit

# 子依赖
async def get_current_user(token: str = Depends(oauth2_scheme)):
    return decode_token(token)

async def get_admin_user(user: User = Depends(get_current_user)):
    if not user.is_admin:
        raise HTTPException(403)
    return user

# 异步路由（推荐用于 I/O 密集）
@app.get("/async")
async def async_endpoint():
    data = await fetch_from_database()
    return data

# 同步路由（自动在线程池执行）
@app.get("/sync")
def sync_endpoint():
    data = blocking_operation()  # CPU 密集或遗留同步代码
    return data

from typing import List
from pydantic import Field

class Item(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0)
    tags: List[str] = Field(default=[], max_items=10)

@app.post(
    "/items/",
    response_model=Item,                      # 响应模型
    status_code=201,                          # 默认状态码
    tags=["items"],                           # OpenAPI 标签
    summary="Create an item",                 # 摘要
    response_description="The created item",  # 响应描述
)
async def create_item(item: Item):
    """
    Create an item with all the information:
    
    - **name**: item name (1-100 characters)
    - **price**: item price (must be positive)
    - **tags**: optional tags (max 10)
    """
    return item

from fastapi import FastAPI

# 创建 FastAPI 应用实例
app = FastAPI()

@app.get("/")
async def read_root():
    """根路径处理器，返回欢迎消息
    
    Returns:
        dict: 包含欢迎消息的字典
    """
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    """获取单个项目的详细信息
    
    Args:
        item_id (int): 项目的唯一标识符
        q (str, optional): 查询字符串参数
        
    Returns:
        dict: 包含项目ID和查询参数的字典
    """
    return {"item_id": item_id, "q": q}

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

app = FastAPI(
    title="用户管理API",
    description="一个完整的用户管理系统API示例",
    version="1.0.0"
)

# Pydantic 数据模型定义
class User(BaseModel):
    """用户数据模型"""
    id: Optional[int] = None
    username: str
    email: str
    full_name: Optional[str] = None
    disabled: bool = False
    created_at: Optional[datetime.datetime] = None

class UserCreate(BaseModel):
    """创建用户请求模型"""
    username: str
    email: str
    full_name: Optional[str] = None
    password: str

# 模拟数据库
fake_users_db = []

@app.post("/users/", response_model=User)
async def create_user(user: UserCreate):
    """创建新用户
    
    Args:
        user (UserCreate): 包含用户信息的请求体
        
    Returns:
        User: 创建成功的用户信息（不含密码）
        
    Raises:
        HTTPException: 当用户名已存在时抛出409错误
    """
    # 检查用户名是否已存在
    for existing_user in fake_users_db:
        if existing_user["username"] == user.username:
            raise HTTPException(status_code=409, detail="用户名已存在")
    
    # 创建新用户
    new_user = {
        "id": len(fake_users_db) + 1,
        "username": user.username,
        "email": user.email,
        "full_name": user.full_name,
        "disabled": False,
        "created_at": datetime.datetime.now()
    }
    fake_users_db.append(new_user)
    
    return User(**new_user)

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

app = FastAPI()
security = HTTPBearer()

# 模拟token验证
fake_tokens = {
    "super_secret_token": {"username": "admin", "permissions": ["read", "write"]},
    "user_token": {"username": "user", "permissions": ["read"]}
}

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """验证访问令牌"""
    token = credentials.credentials
    if token not in fake_tokens:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的访问令牌",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return fake_tokens[token]

@app.get("/protected")
async def protected_route(current_user: dict = Depends(verify_token)):
    """受保护的路由，需要身份验证"""
    return {"message": f"Hello, {current_user['username']}!", "data": "受保护的数据"}