# FastAPI 快速参考

*路径操作、验证、依赖注入、认证、测试*

> Source: FastAPI Documentation (fastapi.tiangolo.com) · MIT

## 初始化

### 最简应用

```
from fastapi import FastAPI
app = FastAPI()

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

### 运行应用

```
pip install "fastapi[standard]"
fastapi dev main.py    # dev with auto-reload
fastapi run main.py    # production
```

### 核心特性

| Command | Description |
|---------|-------------|
| `Async native` | async/await，ASGI（Uvicorn） |
| `Auto docs` | Swagger UI 在 `/docs`，ReDoc 在 `/redoc` |
| `Type validation` | Pydantic 模型验证请求/响应 |
| `OpenAPI` | 自动生成 OpenAPI schema |
| `Dependency injection` | 内置依赖注入系统 |

## 路径操作

### HTTP 方法

```
@app.get("/items")
@app.post("/items")
@app.put("/items/{item_id}")
@app.patch("/items/{item_id}")
@app.delete("/items/{item_id}")
```

### 路径参数

```
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

# Enum constraint
from enum import Enum
class Color(str, Enum):
    red = "red"
    blue = "blue"
```

### 状态码与标签

```
from fastapi import status

@app.post("/items", status_code=status.HTTP_201_CREATED,
          tags=["items"])
async def create_item(item: Item):
    return item
```

## 请求体

### Pydantic 模型

```
from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str
    price: float = Field(gt=0, description="Must be positive")
    tags: list[str] = []
```

### 嵌套模型

```
class Address(BaseModel):
    street: str
    city: str
    zip_code: str

class User(BaseModel):
    name: str
    address: Address
```

### 在端点中使用

```
@app.post("/items")
async def create_item(item: Item):
    return {"name": item.name, "price": item.price}
```

### 验证特性

| Command | Description |
|---------|-------------|
| `Field(gt=0)` | 大于 0 |
| `Field(min_length=1)` | 最小字符串长度 |
| `Field(max_length=100)` | 最大字符串长度 |
| `Field(pattern='^[a-z]+$')` | 正则模式匹配 |
| `Field(default=None)` | 带默认值的可选字段 |
| `EmailStr` | 邮箱验证（pydantic[email]） |

## 查询参数

### 基础查询参数

```
@app.get("/items")
async def list_items(skip: int = 0, limit: int = 10):
    return items[skip : skip + limit]
# GET /items?skip=0&limit=20
```

### 查询验证

```
from fastapi import Query

@app.get("/search")
async def search(
    q: str = Query(min_length=3, max_length=50),
    page: int = Query(default=1, ge=1),
):
    return {"q": q, "page": page}
```

### 可选与必填

```
async def read_items(
    q: str | None = None,   # optional
    name: str = ...,         # required (Ellipsis)
    tags: list[str] = Query(default=[]),
):
    return {"q": q, "name": name}
```

### Header 与 Cookie

```
from fastapi import Header, Cookie

async def read(
    user_agent: str | None = Header(default=None),
    session_id: str | None = Cookie(default=None),
):
    return {"ua": user_agent}
```

## 响应模型

### 响应模型

```
class ItemOut(BaseModel):
    name: str
    price: float

@app.get("/items/{id}", response_model=ItemOut)
async def get_item(id: int):
    return items[id]  # filters out extra fields
```

### 多种响应类型

```
from fastapi.responses import JSONResponse, HTMLResponse

@app.get("/html", response_class=HTMLResponse)
async def get_html():
    return "<h1>Hello</h1>"
```

### 响应模型选项

| Command | Description |
|---------|-------------|
| `response_model` | 用于输出过滤的 Pydantic 模型 |
| `response_model_exclude_unset` | 省略未显式设置的字段 |
| `response_model_include` | 白名单特定字段 |
| `response_model_exclude` | 黑名单特定字段 |

### 错误响应

```
from fastapi import HTTPException

@app.get("/items/{id}")
async def get_item(id: int):
    if id not in items:
        raise HTTPException(status_code=404, detail="Not found")
    return items[id]
```

## 依赖注入

### 函数依赖

```
from fastapi import Depends

async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
```

### 在端点中使用

```
@app.get("/users")
async def list_users(db: Session = Depends(get_db)):
    return db.query(User).all()
```

### 类依赖

```
class Pagination:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip = skip
        self.limit = limit

@app.get("/items")
async def list_items(pg: Pagination = Depends()):
    return items[pg.skip : pg.skip + pg.limit]
```

### 依赖作用域

| Command | Description |
|---------|-------------|
| `Depends(func)` | 端点级依赖 |
| `app = FastAPI(dependencies=[...])` | 所有路由的全局依赖 |
| `APIRouter(dependencies=[...])` | Router 级依赖 |
| `yield` | 安装/拆卸（DB session、锁） |

## 认证

### OAuth2 密码 Bearer

```
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/users/me")
async def read_me(token: str = Depends(oauth2_scheme)):
    user = decode_token(token)
    return user
```

### JWT Token 流程

```
from jose import jwt
SECRET = "your-secret-key"

def create_token(data: dict):
    return jwt.encode(data, SECRET, algorithm="HS256")

def decode_token(token: str):
    return jwt.decode(token, SECRET, algorithms=["HS256"])
```

### Token 端点

```
from fastapi.security import OAuth2PasswordRequestForm

@app.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends()):
    user = authenticate(form.username, form.password)
    if not user:
        raise HTTPException(status_code=401)
    return {"access_token": create_token({"sub": user.id})}
```

### 安全方案

| Command | Description |
|---------|-------------|
| `OAuth2PasswordBearer` | 通过表单登录的 Bearer token |
| `HTTPBasic` | 基础用户名/密码认证 |
| `APIKeyHeader` | Header 中的 API key |
| `APIKeyCookie` | Cookie 中的 API key |

## 后台任务

### 简单后台任务

```
from fastapi import BackgroundTasks

def send_email(to: str, body: str):
    # slow operation runs after response
    email_client.send(to, body)

@app.post("/notify")
async def notify(bg: BackgroundTasks):
    bg.add_task(send_email, "user@example.com", "Hello!")
    return {"status": "queued"}
```

### 依赖中使用后台任务

```
async def log_request(bg: BackgroundTasks):
    bg.add_task(write_log, "request received")

@app.get("/items", dependencies=[Depends(log_request)])
async def list_items():
    return items
```

### 后台任务 vs Worker

| Command | Description |
|---------|-------------|
| `BackgroundTasks` | 响应后的轻量任务（邮件、日志） |
| `Celery / ARQ` | 需要独立 worker 的重任务 |
| `asyncio.create_task` | 即发即忘的异步协程 |

## 中间件

### 自定义中间件

```
import time
from starlette.middleware.base import BaseHTTPMiddleware

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start
        response.headers["X-Process-Time"] = str(duration)
        return response
```

### 添加中间件

```
app.add_middleware(TimingMiddleware)
```

### CORS

```
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_methods=["*"],
    allow_headers=["*"],
)
```

### 内置中间件

| Command | Description |
|---------|-------------|
| `CORSMiddleware` | 跨域资源共享 |
| `TrustedHostMiddleware` | 限制允许的主机名 |
| `GZipMiddleware` | Gzip 响应压缩 |
| `HTTPSRedirectMiddleware` | HTTP 重定向到 HTTPS |

## 测试

### 测试客户端

```
from fastapi.testclient import TestClient

client = TestClient(app)

def test_read_root():
    resp = client.get("/")
    assert resp.status_code == 200
    assert resp.json() == {"message": "Hello, World!"}
```

### 测试 POST

```
def test_create_item():
    resp = client.post("/items", json={
        "name": "Widget",
        "price": 9.99,
    })
    assert resp.status_code == 201
    assert resp.json()["name"] == "Widget"
```

### 覆盖依赖

```
async def mock_db():
    return FakeDB()

app.dependency_overrides[get_db] = mock_db

def test_with_mock_db():
    resp = client.get("/users")
    assert resp.status_code == 200
```

### 异步测试

```
import pytest
from httpx import AsyncClient, ASGITransport

@pytest.mark.anyio
async def test_async():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport) as ac:
        resp = await ac.get("/")
    assert resp.status_code == 200
```