FastAPI 全面指南

FastAPI 是一个现代、快速（高性能）的 Python Web 框架，基于标准 Python 类型提示构建。它由 Sebastián Ramírez 於 2018 年创建，旨在结合现代 Python 特性与 Web 开发最佳实践，提供开发效率与运行性能兼具的解决方案。

核心亮点：FastAPI 是最快的 Python Web 框架之一，性能可与 NodeJS、Go 相媲美。

一、FastAPI 概述

1.1 核心特性

• 快速编码：开发速度提升约 200%~300%
• 更少 Bug：减少约 40% 的人为错误
• 直观简洁：强大的编辑器支持，无处不在的自动补全
• 易于调试：Less time debugging
• 基于标准：完全兼容 OpenAPI（原 Swagger）和 JSON Schema
• 简洁代码：最小化代码重复
• 稳健可靠：生产就绪的代码，自动交互式文档
• 标准化：基于并完全兼容 API 的开放标准

1.2 技术栈

FastAPI 构建在与 Python 社区广泛使用的基础工具之上：

1.3 与其他框架对比

二、安装与快速开始

2.1 安装

pip install fastapi
pip install "uvicorn[standard]"

2.2 第一个应用

创建 main.py：

from fastapi import FastAPI

app = FastAPI()

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

启动开发服务器：

uvicorn main:app --reload

参数说明：

• main：文件名 main.py
• app：应用实例名 app = FastAPI()
• --reload：代码修改后自动重启（仅开发环境使用）

2.3 查看交互式文档

FastAPI 自动生成交互式 API 文档，无需额外配置：

• Swagger UI：http://127.0.0.1:8000/docs
• ReDoc：http://127.0.0.1:8000/redoc

这两个页面可以直接在浏览器中测试 API 接口，是开发调试的利器。

三、路径参数

3.1 基本用法

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}

访问 /items/foo 返回 {"item_id": "foo"}。

3.2 类型检查

FastAPI 使用 Python 类型提示自动进行参数类型转换和验证：

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

当访问 /items/3 时，item_id 会被自动转换为整数 3。若访问 /items/abc，FastAPI 返回清晰的错误信息：

{
    "detail": [
        {
            "type": "int_parsing",
            "loc": ["path", "item_id"],
            "msg": "Input should be a valid integer"
        }
    ]
}

3.3 预定义路径值

使用 Enum 限定路径参数的取值范围：

from enum import Enum
from fastapi import FastAPI

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

app = FastAPI()

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    if model_name is ModelName.alexnet:
        return {"model": model_name, "message": "Deep Learning"}
    return {"model": model_name}

3.4 路径包含文件路径

当路径参数本身包含斜杠时，使用 Starlette 路径转换器：

@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

访问 /files/home/john/file.txt 时，file_path 为 "home/john/file.txt"。

四、查询参数

4.1 基本查询参数

不在路径中的参数会自动识别为查询参数：

from fastapi import FastAPI

app = FastAPI()

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip : skip + limit]

访问 /items/?skip=0&limit=10 即可获取分页数据。

4.2 可选参数与默认值

from typing import Optional
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Optional[str] = None, short: bool = False):
    item = {"item_id": item_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update({"description": "This is a long description"})
    return item

4.3 必需查询参数

未设置默认值的参数为必需参数：

@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str):
    return {"item_id": item_id, "needy": needy}

访问 /items/1 缺少 needy 时返回 422 错误。

布尔值自动转换：short=1、short=True、short=true、short=on、short=yes 都会被转换为布尔值 True。

五、请求体

5.1 定义数据模型

使用 Pydantic 的 BaseModel 定义请求体结构：

from pydantic import BaseModel
from fastapi import FastAPI

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

@app.post("/items/")
async def create_item(item: Item):
    return item

FastAPI 自动完成以下工作：

1. 将请求体读取为 JSON
2. 转换字段为对应类型
3. 验证数据，失败时返回清晰的 422 错误
4. 将数据封装为 item 参数传入函数

5.2 使用模型

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.model_dump()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

Pydantic v2 使用 model_dump() 方法将模型转换为字典（v1 中为 dict()）。

5.3 请求体 + 路径参数 + 查询参数

FastAPI 能自动识别参数来源：

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: Optional[str] = None):
    result = {"item_id": item_id, **item.model_dump()}
    if q:
        result.update({"q": q})
    return result

• 路径中的 item_id → 路径参数
• Pydantic 模型 item → 请求体
• 简单类型 q → 查询参数

5.4 嵌套模型

Pydantic 支持任意深度的嵌套模型：

from typing import List
from pydantic import BaseModel

class Image(BaseModel):
    url: str
    name: str

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: List[str] = []
    image: Optional[Image] = None

@app.post("/items/")
async def create_item(item: Item):
    return item

请求体示例：

{
    "name": "Foo",
    "description": "A nice item",
    "price": 45.2,
    "tax": 3.2,
    "tags": ["x", "y"],
    "image": {
        "url": "http://example.com/img.png",
        "name": "Image"
    }
}

5.5 List 类型的请求体

可以使用嵌入模型列表作为请求体：

from typing import List
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items/bulk")
async def create_items(items: List[Item]):
    return items

六、参数验证

6.1 字符串验证

使用 Query、Path、Field 等对象为参数添加验证约束：

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    q: Optional[str] = Query(
        None,
        min_length=3,
        max_length=50,
        regex="^fixedquery$"
    )
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

验证参数说明：

6.2 数值验证

使用 Path 对象验证路径参数中的数值：

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_items(
    item_id: int = Path(..., title="Item ID", ge=1, le=1000),
    q: Optional[str] = Query(None, alias="item-query")
):
    return {"item_id": item_id, "q": q}

数值验证条件：

• gt：大于
• ge：大于等于
• lt：小于
• le：小于等于

6.3 使用 Field 添加验证

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    description: Optional[str] = Field(None, max_length=300)
    price: float = Field(..., gt=0, description="价格必须大于0")

@app.post("/items/")
async def create_item(item: Item):
    return item

6.4 Body 验证

通过 Body 可以对请求体附加额外验证：

from typing import Optional
from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, importance: int = Body(gt=1)):
    return {"item_id": item_id, "item": item, "importance": importance}

请求体格式变为：

{
    "item": {"name": "Foo", "description": "..."},
    "importance": 5
}

七、响应模型

7.1 response_model 参数

使用 response_model 确保响应数据的结构一致性，避免暴露敏感字段：

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

app = FastAPI()

class User(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None
    password: str  # 敏感字段

class UserOut(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None

@app.post("/users/", response_model=UserOut)
async def create_user(user: User):
    return user

尽管返回了完整的 User 对象，FastAPI 会根据 response_model=UserOut 过滤掉 password 字段。

7.2 响应模型默认值

在响应模型中可以直接定义默认值：

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: float = 10.5

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The Bar fighters", "price": 62, "tax": 20.2},
}

@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items.get(item_id)

响应模型控制选项：

7.3 列表响应

@app.get("/items/", response_model=List[Item])
async def read_items():
    return [
        {"name": "Foo", "price": 50.2},
        {"name": "Bar", "price": 62.0},
    ]

八、异步编程

8.1 async / await

FastAPI 原生支持 Python 的 async / await 语法：

import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/async-endpoint")
async def read_async():
    await asyncio.sleep(1)
    return {"message": "异步处理完成"}

8.2 异步数据库调用示例

import asyncio
from fastapi import FastAPI

app = FastAPI()

async def fake_db_query():
    await asyncio.sleep(1)
    return {"name": "Foo", "price": 50.2}

@app.get("/items/")
async def read_items():
    item = await fake_db_query()
    return item

8.3 同步路由处理

对于不使用异步的函数（如不支持 await 的库），直接使用普通函数即可。FastAPI 会自动在线程池中运行它：

@app.get("/sync-endpoint")
def read_sync():
    time.sleep(1)  # 同步阻塞调用
    return {"message": "同步处理完成"}

重要决策原则：如果函数内部调用了 async def 的协程，使用 async def。如果调用了不支持异步的同步阻塞代码（如某些数据库驱动），使用普通 def，FastAPI 会在线程池中运行避免阻塞事件循环。

8.4 并发请求

import asyncio
from fastapi import FastAPI

app = FastAPI()

async def fetch_data(url: str):
    await asyncio.sleep(1)
    return {"data": f"from {url}"}

@app.get("/dashboard")
async def get_dashboard():
    results = await asyncio.gather(
        fetch_data("api_1"),
        fetch_data("api_2"),
        fetch_data("api_3"),
    )
    return {"results": results}

使用 asyncio.gather() 并发执行多个异步调用，总耗时约为最长单个请求的时间（1 秒），而非串行的 3 秒。

九、依赖注入

FastAPI 内置强大的依赖注入系统，支持类型化的组件和依赖树。

9.1 基本用法

from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return {"message": "Hello", **commons}

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return {"message": "Hello Users", **commons}

依赖注入的优势：

• 代码复用（共享逻辑）
• 数据库连接共享（session）
• 认证和角色验证
• 避免代码重复

9.2 类作为依赖

from fastapi import Depends, FastAPI

app = FastAPI()

class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

FastAPI 支持简写形式，以下三种写法等价：

commons = Depends(CommonQueryParams)
commons = Depends()  # 简写，FastAPI 自动推断类型
commons: CommonQueryParams  # 仅类型提示，FastAPI 自动注入

9.3 子依赖

依赖可以有自己的依赖，形成依赖树：

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()

def query_extractor(q: Optional[str] = None):
    return q

def query_or_cookie_extractor(
    q: str = Depends(query_extractor),
    last_query: Optional[str] = Cookie(None)
):
    if not q:
        return last_query
    return q

@app.get("/items/")
async def read_query(orb: str = Depends(query_or_cookie_extractor)):
    return {"q_or_cookie": orb}

9.4 依赖缓存与 use_cache

默认情况下，同一请求内多次引入同一依赖会被缓存（返回相同值）。禁用缓存：

@app.get("/items/")
async def read_items(
    commons: dict = Depends(common_parameters, use_cache=False)
):
    return commons

9.5 Yield 依赖（资源管理）

使用 yield 实现依赖的前置和清理逻辑：

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

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

相当于 Python 的 with 上下文管理器，在请求结束时自动关闭数据库连接。

9.6 全局依赖

可以在整个应用或路由组上应用依赖：

async def verify_token(x_token: str = Header()):
    if x_token != "supersecrettoken":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

app = FastAPI(dependencies=[Depends(verify_token)])

@app.get("/protected")
async def protected():
    return {"message": "Access granted"}

9.7 基于 Path 的依赖

@app.get("/items/{item_id}", dependencies=[Depends(verify_token)])
async def read_item(item_id: int):
    return {"item_id": item_id}

十、安全与认证

FastAPI 集成了 fastapi.security 模块，提供多种安全方案。

10.1 OAuth2 密码流

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

fake_users_db = {
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonder",
        "email": "alice@example.com",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    }
}

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

class UserInDB(User):
    hashed_password: str

def fake_hash_password(password: str):
    return "fakehashed" + password

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(form_data.username)
    if not user:
        raise HTTPException(status_code=400, detail="用户名或密码错误")
    hashed_password = fake_hash_password(form_data.password)
    if hashed_password != user["hashed_password"]:
        raise HTTPException(status_code=400, detail="用户名或密码错误")
    return {"access_token": user["username"], "token_type": "bearer"}

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = fake_users_db.get(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的认证凭据",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return UserInDB(**user)

async def get_current_active_user(current_user: UserInDB = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="用户已被禁用")
    return current_user

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

10.2 JWT Token

结合 python-jose 生成 JWT Token：

from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt

SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

def decode_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

10.3 API Key 认证

from fastapi import Security
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

async def get_api_key(api_key: str = Security(api_key_header)):
    if api_key != "my-secret-key":
        raise HTTPException(status_code=403, detail="无效的 API Key")
    return api_key

@app.get("/secure-data")
async def secure_data(api_key: str = Depends(get_api_key)):
    return {"data": "这是受保护的数据"}

10.4 密码哈希

使用 passlib 进行安全的密码哈希：

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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

def get_password_hash(password):
    return pwd_context.hash(password)

10.5 HTTP Basic Auth

from fastapi import Depends, FastAPI
from fastapi.security import HTTPBasic, HTTPBasicCredentials

app = FastAPI()
security = HTTPBasic()

@app.get("/users/me")
async def get_current_user(credentials: HTTPBasicCredentials = Depends(security)):
    correct_username = "admin"
    correct_password = "secret"
    if credentials.username != correct_username or credentials.password != correct_password:
        raise HTTPException(
            status_code=401,
            detail="认证失败",
            headers={"WWW-Authenticate": "Basic"},
        )
    return {"username": credentials.username}

10.6 CORS 中间件

跨域资源共享配置：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://dailyblue.cloud", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

十一、中间件

中间件是在请求到达路由之前和响应返回之前执行的函数。

11.1 自定义中间件

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

11.2 使用 BaseHTTPMiddleware

from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 请求前处理
        response = await call_next(request)
        # 响应后处理
        response.headers["Custom-Header"] = "Custom Value"
        return response

app = FastAPI()
app.add_middleware(CustomMiddleware)

11.3 HTTPS 重定向

from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()
app.add_middleware(HTTPSRedirectMiddleware)

11.4 受信任主机

from fastapi.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["dailyblue.cloud", "*.dailyblue.cloud", "localhost"],
)

11.5 GZip 压缩

from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000)

十二、后台任务

12.1 BackgroundTasks

FastAPI 内置了轻量级的后台任务支持：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message: str):
    with open("log.txt", mode="a") as email_file:
        content = f"notification for {email}: {message}\n"
        email_file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, "some message")
    return {"message": "Notification sent in background"}

BackgroundTasks 的特点：

• 与依赖注入系统兼容
• 任务按添加顺序执行
• 适用于发送邮件、写入日志等轻量任务
• 对于 CPU 密集型任务建议使用 Celery

12.2 依赖中的后台任务

 def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)

def get_query(background_tasks: BackgroundTasks, q: Optional[str] = None):
    if q:
        background_tasks.add_task(write_log, f"found query: {q}")
    return q

@app.get("/")
async def read_main(
    q: str = Depends(get_query),
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(write_log, "end of request")
    return {"message": "Hello World"}

十三、WebSocket

FastAPI 支持 WebSocket 连接：

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 the chat")

十四、数据库集成

14.1 SQLAlchemy 集成

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from typing import List

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

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

class DBItem(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String)

Base.metadata.create_all(bind=engine)

app = FastAPI()

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

class ItemCreate(BaseModel):
    title: str
    description: Optional[str] = None

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

@app.post("/items/", response_model=Item)
def create_item(item: ItemCreate, db = Depends(get_db)):
    db_item = DBItem(**item.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.get("/items/", response_model=List[Item])
def read_items(skip: int = 0, limit: int = 100, db = Depends(get_db)):
    items = db.query(DBItem).offset(skip).limit(limit).all()
    return items

14.2 异步数据库

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

ASYNC_DATABASE_URL = "postgresql+asyncpg://user:password@localhost/db"

async_engine = create_async_engine(ASYNC_DATABASE_URL)
AsyncSessionLocal = sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
)

async def get_async_db():
    async with AsyncSessionLocal() as db:
        yield db

14.3 数据库迁移

推荐使用 Alembic 进行数据库迁移管理：

pip install alembic
alembic init alembic

修改 alembic.ini 中的数据库连接字符串，然后创建迁移：

alembic revision --autogenerate -m "create items table"
alembic upgrade head

十五、异常处理

15.1 HTTPException

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "Some custom header"}
        )
    return {"item": items[item_id]}

15.2 自定义异常

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

app = FastAPI()

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops, {exc.name} did it again"},
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

15.3 全局异常处理器

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": "自定义验证错误信息", "errors": exc.errors()},
    )

十六、测试

16.1 TestClient

使用 Starlette 提供的 TestClient 进行测试：

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, FastAPI!"}

def test_read_item():
    response = client.get("/items/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

16.2 pytest fixture

import pytest
from fastapi.testclient import TestClient
from main import app

@pytest.fixture(scope="module")
def client():
    with TestClient(app) as c:
        yield c

def test_create_item(client):
    response = client.post(
        "/items/",
        json={"name": "Foo", "price": 50.2}
    )
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Foo"
    assert data["price"] == 50.2

def test_invalid_item(client):
    response = client.post(
        "/items/",
        json={"name": "Foo"}  # 缺少 price
    )
    assert response.status_code == 422

16.3 测试依赖覆盖

from fastapi.testclient import TestClient
from main import app, get_db

# 替换数据库依赖
def override_get_db():
    try:
        db = TestSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)

def test_read_items():
    response = client.get("/items/")
    assert response.status_code == 200

十七、项目结构与配置

17.1 推荐项目结构

my_project/
├── app/
│   ├── __init__.py
│   ├── main.py          # 应用入口
│   ├── dependencies.py   # 依赖
│   ├── models.py         # Pydantic 模型
│   ├── database.py       # 数据库配置
│   ├── crud.py           # CRUD 操作
│   ├── schemas.py        # 数据验证模型
│   └── routers/
│       ├── __init__.py
│       ├── items.py
│       └── users.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_items.py
├── requirements.txt
└── alembic/

17.2 路由模块化

使用 APIRouter 组织路由：

from fastapi import APIRouter, Depends

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(verify_token)],
    responses={404: {"description": "Not found"}},
)

@router.get("/")
async def read_items():
    return [{"item": "Foo"}]

@router.get("/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

在主应用中注册：

from fastapi import FastAPI
from app.routers import items, users

app = FastAPI()
app.include_router(items.router)
app.include_router(users.router)

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

17.3 配置管理

使用 pydantic-settings 管理配置：

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "Awesome API"
    admin_email: str = "admin@example.com"
    database_url: str = "sqlite:///./test.db"
    secret_key: str = "change-me"

    class Config:
        env_file = ".env"

settings = Settings()

@app.get("/info")
async def info():
    return {
        "app_name": settings.app_name,
        "admin_email": settings.admin_email,
    }

.env 文件：

APP_NAME=My FastAPI App
ADMIN_EMAIL=admin@dailyblue.cloud
DATABASE_URL=postgresql://user:pass@localhost/db
SECRET_KEY=super-secret-key

十八、启动与关闭事件

18.1 lifespan 事件

FastAPI 推荐使用 lifespan 代替已弃用的 @app.on_event：

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行
    print("应用启动中...")
    db = create_db_connection()
    app.state.db = db
    yield
    # 关闭时执行
    print("应用关闭中...")
    await db.close()

app = FastAPI(lifespan=lifespan)

18.2 启动任务

from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 加载 ML 模型
    app.state.model = load_model()
    # 初始化缓存
    app.state.cache = await create_redis_pool()
    yield
    # 清理资源
    await app.state.cache.close()

app = FastAPI(lifespan=lifespan)

十九、静态文件与文件上传

19.1 静态文件服务

from fastapi.staticfiles import StaticFiles

app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")

19.2 文件上传

from fastapi import FastAPI, UploadFile, File
from typing import List

app = FastAPI()

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(contents),
    }

@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}

19.3 保存上传文件

import shutil
from pathlib import Path

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    upload_dir = Path("uploads")
    upload_dir.mkdir(exist_ok=True)

    file_path = upload_dir / file.filename
    with file_path.open("wb") as buffer:
        shutil.copyfileobj(file.file, buffer)

    return {"filename": file.filename, "saved_to": str(file_path)}

二十、表单与 Cookie

20.1 表单数据

from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username}

需要安装 python-multipart：

pip install python-multipart

20.2 Cookie 读取

from fastapi import Cookie, FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(ads_id: Optional[str] = Cookie(None)):
    return {"ads_id": ads_id}

20.3 Header 读取

from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(
    user_agent: Optional[str] = Header(None),
    x_token: Optional[str] = Header(None, alias="X-Token")
):
    return {"User-Agent": user_agent, "X-Token": x_token}

自动转换：FastAPI 会自动将参数名中的下划线转换为 Header 中的连字符。例如 user_agent 会匹配 User-Agent 头。

二十一、流式响应与其他响应类型

21.1 StreamingResponse

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time

app = FastAPI()

def generate_data():
    for i in range(10):
        yield f"数据行 {i}\n"
        time.sleep(0.5)

@app.get("/stream")
async def stream_data():
    return StreamingResponse(generate_data(), media_type="text/plain")

21.2 JSONResponse

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/custom-json")
async def custom_json():
    return JSONResponse(
        status_code=200,
        content={"message": "自定义响应", "code": 200},
        headers={"X-Custom-Header": "value"}
    )

21.3 HTMLResponse

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.get("/html", response_class=HTMLResponse)
async def read_html():
    return """
    <html>
        <head><title>FastAPI HTML</title></head>
        <body><h1>Hello, FastAPI!</h1></body>
    </html>
    """

21.4 FileResponse

from fastapi import FastAPI
from fastapi.responses import FileResponse

app = FastAPI()

@app.get("/download")
async def download_file():
    return FileResponse(
        path="files/report.pdf",
        filename="report.pdf",
        media_type="application/pdf"
    )

21.5 RedirectResponse

from fastapi import FastAPI
from fastapi.responses import RedirectResponse

app = FastAPI()

@app.get("/old-path")
async def old_path():
    return RedirectResponse(url="/new-path")

@app.get("/new-path")
async def new_path():
    return {"message": "新路径"}

二十二、自定义 OpenAPI 文档

22.1 应用元信息

from fastapi import FastAPI

app = FastAPI(
    title="My Awesome API",
    description="这是我的 awesome API 文档",
    version="1.0.0",
    terms_of_service="https://dailyblue.cloud/terms/",
    contact={
        "name": "Dailyblue",
        "url": "https://dailyblue.cloud",
        "email": "admin@dailyblue.cloud",
    },
    license_info={
        "name": "Apache 2.0",
        "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
    },
)

22.2 文档 URL 地址与禁用

app = FastAPI(
    docs_url="/documentation",
    redoc_url=None,  # 禁用 ReDoc
    openapi_url="/custom-openapi.json"
)

22.3 OpenAPI Schema 自定义

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    openapi_schema = get_openapi(
        title="Custom API",
        version="1.0.0",
        description="API description",
        routes=app.routes,
    )
    # 可以在这里修改或扩展 schema
    openapi_schema["info"]["x-custom"] = "Custom value"
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

二十三、部署

23.1 Uvicorn 部署

生产环境部署（单进程）：

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

参数说明：

• --host：绑定地址
• --port：端口
• --workers：工作进程数（建议设为 CPU 核心数的 2-4 倍）
• 不使用 --reload（生产关闭热重载）

23.2 Gunicorn + Uvicorn worker

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

安装：

pip install gunicorn uvicorn

23.3 Docker 部署

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

构建运行：

docker build -t myfastapi .
docker run -d -p 8000:8000 myfastapi

23.4 Nginx 反向代理

server {
    listen 80;
    server_name api.dailyblue.cloud;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }

    location /ws {
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

23.5 Systemd 服务

[Unit]
Description=FastAPI Application
After=network.target

[Service]
User=ubuntu
Group=ubuntu
WorkingDirectory=/home/ubuntu/myapi
Environment="PATH=/home/ubuntu/myapi/venv/bin"
ExecStart=/home/ubuntu/myapi/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8000
Restart=always
RestartSec=3

[Install]
WantedBy=multi-user.target

二十四、性能优化

24.1 异步 I/O 利用

充分利用异步 I/O 提升吞吐量：

import httpx
import asyncio

async def fetch_multiple(urls: list):
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        return await asyncio.gather(*tasks)

@app.get("/aggregate")
async def aggregate():
    urls = ["https://api1.com", "https://api2.com", "https://api3.com"]
    responses = await fetch_multiple(urls)
    return [r.json() for r in responses]

24.2 缓存策略

from functools import lru_cache

@lru_cache()
def get_settings():
    return Settings()

@app.get("/config")
async def get_config():
    return get_settings()

24.3 连接池与资源复用

from databases import Database

database = Database("postgresql://user:pass@localhost/db")

@asynccontextmanager
async def lifespan(app: FastAPI):
    await database.connect()
    yield
    await database.disconnect()

app = FastAPI(lifespan=lifespan)

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    query = "SELECT * FROM users WHERE id = :user_id"
    return await database.fetch_one(query, values={"user_id": user_id})

二十五、FastAPI 最佳实践

25.1 代码组织

• 路由按业务模块拆分到 routers/ 目录
• 数据模型、数据库模型、Schema 分别定义
• 业务逻辑放在独立的 services/ 或 crud.py
• 配置使用 pydantic-settings 统一管理

25.2 性能建议

• 使用 async def 配合异步库（httpx、asyncpg、databases）
• 使用 def 处理同步阻塞调用（FastAPI 会自动放入线程池）
• 合理使用缓存（lru_cache、Redis）
• 数据库查询注意 N+1 问题
• 生产环境设置合理的 workers 数量

25.3 安全建议

• 使用 HTTPS
• 实现 CORS 白名单
• 使用 JWT + OAuth2 进行认证
• 对敏感配置使用环境变量
• 开启请求速率限制
• 避免在错误信息中暴露内部细节

25.4 错误处理建议

• 使用统一的响应格式
• 定义全局异常处理器
• 记录请求日志
• 使用结构化日志（structlog 或 logging）

25.5 测试建议

• 为每个接口编写测试
• 使用 dependency_overrides 替换数据库依赖
• 测试异常路径
• 使用 pytest + pytest-asyncio

二十六、常用快捷工具

26.1 生成 requirements.txt

pip freeze > requirements.txt

26.2 FastAPI CLI

FastAPI 提供了 CLI 工具直接运行应用：

fastapi dev main.py          # 开发模式
fastapi run main.py          # 生产模式
fastapi run --workers 4 main.py

26.3 常用第三方扩展

总结

FastAPI 凭借以下优势，已成为 Python Web 开发的首选框架之一：

1. 类型驱动开发：基于 Python 类型提示，自动完成验证、转换、文档生成
2. 高性能异步：原生 async/await 支持，性能媲美 NodeJS 和 Go
3. 开发效率：自动交互式文档，减少手写 boilerplate
4. 生产就绪：完善的依赖注入、安全、数据库集成支持
5. 活跃生态：Starlette + Pydantic 社区支持，扩展丰富

一句话总结：FastAPI = Starlette（路由）+ Pydantic（验证）+ asyncio（性能）+ 自动文档

无论是构建 RESTful API、WebSocket 服务、微服务后端还是实时应用，FastAPI 都提供了清晰、高效、现代的开发体验。对于 Python 开发者来说，它值得作为构建 Web 服务的首选框架。