FastAPI：现代高性能 Python Web 框架

使用自动文档和类型安全构建超快速的 API

目录

FastAPI 已经成为构建 API 最令人兴奋的 Python Web 框架之一，结合了现代 Python 特性、卓越的性能和开发体验。

无论您是构建微服务、无服务器函数还是复杂的 Web 应用程序，FastAPI 都能为您提供构建生产就绪 API 所需的工具。

python 和 fastapi 城堡

什么是 FastAPI？

FastAPI 是一个现代、高性能的 Web 框架，用于使用 Python 3.8+ 构建 API，基于标准 Python 类型提示。由 Sebastián Ramírez 创建，它被设计为易于使用、编码速度快，并且从一开始就具备生产就绪性。

该框架依托于两个强大的库：

  • Starlette 用于 Web 路由和异步功能
  • Pydantic 用于使用 Python 类型提示进行数据验证

这种组合提供了与 Node.js 和 Go 相当的性能，同时保持了 Python 的简单性和可读性。FastAPI 迅速获得了关注，在 GitHub 上获得了数千个星标，并被全球各大公司采用。

使 FastAPI 独树一帜的关键特性

1. 自动 API 文档

FastAPI 最受欢迎的特性之一是自动、交互式 API 文档。基于 OpenAPI 标准（以前称为 Swagger），FastAPI 生成两个文档接口：

  • Swagger UI 在 /docs - 交互式文档，您可以在浏览器中直接测试 API 端点
  • ReDoc 在 /redoc - 以清晰的三面板设计为特色的替代文档

文档是根据代码中的类型提示和文档字符串自动生成的，无需额外配置。这意味着您的文档始终与代码保持同步。

2. 类型提示和验证

FastAPI 利用 Python 的类型提示不仅用于文档，还用于运行时验证和序列化。当您将端点参数定义为 int 时，FastAPI 会自动：

  • 验证请求中是否包含整数
  • 将值从字符串转换为整数
  • 如果验证失败，返回清晰的错误消息
  • 在 API 文档中记录参数类型
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    username: str
    email: str
    age: int

@app.post("/users/")
async def create_user(user: User):
    return {"message": f"User {user.username} created successfully"}

这种方法可以早期捕获错误，减少样板代码，并使您的 API 自我文档化。

3. 异步支持

FastAPI 是基于 ASGI（异步服务器网关接口）构建的，而不是 WSGI，因此它原生支持 async/await。这对于需要进行大量 I/O 调用（数据库查询、API 请求、文件操作）的高性能应用至关重要。

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # 异步操作不会阻塞其他请求
    data = await fetch_from_database(item_id)
    return data

异步支持意味着 FastAPI 可以高效处理数千个并发连接，使其非常适合现代云应用和微服务架构。

4. 性能

FastAPI 是目前最快的 Python 框架之一，其性能与 Node.js 和 Go 框架相当。独立的基准测试一致显示，FastAPI 在 I/O 密集型工作负载上显著优于传统的 Flask 和 Django 框架。

速度来源于：

  • Pydantic 的 C 级验证
  • Starlette 的高效异步实现
  • 框架本身的最小开销

快速入门 FastAPI

安装 FastAPI 非常简单。在使用 Python 项目时，拥有一个合适的 Python 包管理器 可以使依赖管理变得更加容易。您需要 FastAPI 和一个 ASGI 服务器，如 Uvicorn：

pip install fastapi uvicorn[standard]

这是一个最小的 FastAPI 应用程序：

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

访问 http://localhost:8000/docs 查看您的交互式 API 文档。

构建生产就绪的 API

请求和响应模型

为可维护的 API 定义清晰的数据模型至关重要。遵循 Python 清洁架构的设计模式 有助于创建结构良好的应用程序：

from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=8)

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    created_at: datetime
    
    class Config:
        orm_mode = True  # 允许 Pydantic 与 ORM 对象一起工作

@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
    # 您的业务逻辑在这里
    new_user = save_user_to_database(user)
    return new_user

response_model 参数确保 FastAPI 仅返回指定的字段，自动过滤敏感数据，如密码。

依赖注入

FastAPI 的依赖注入系统强大而优雅。它允许您：

  • 在端点之间共享代码
  • 强制执行身份验证/授权
  • 管理数据库连接
  • 处理复杂依赖项
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = verify_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials"
        )
    return user

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

错误处理

FastAPI 提供了内置的异常处理程序，并允许自定义错误响应：

from fastapi import HTTPException

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

背景任务

对于不需要在返回响应前完成的操作：

from fastapi import BackgroundTasks

def send_email_notification(email: str, message: str):
    # 发送邮件逻辑
    pass

@app.post("/send-notification/")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(send_email_notification, email, "Welcome!")
    return {"message": "Notification scheduled"}

测试 FastAPI 应用程序

Python 单元测试 对于可靠 API 至关重要。FastAPI 与 pytest 无缝协作，并提供了一个测试客户端：

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 World"}

def test_read_item():
    response = client.get("/items/42?q=test")
    assert response.status_code == 200
    assert response.json() == {"item_id": 42, "q": "test"}

对于测试异步端点：

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_async_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/")
    assert response.status_code == 200

数据库集成

FastAPI 与各种数据库和 ORM 配合良好。以下是一个使用 SQLAlchemy 的示例：

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

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

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

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

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

部署选项

Docker 部署

FastAPI 应用程序易于容器化。这是一个生产就绪的 Dockerfile 示例：

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"]

AWS Lambda 部署

FastAPI 与无服务器平台配合良好。当 使用 Python 和 Terraform 构建双模式 AWS Lambda 时，可以使用 Mangum，这是一个将 FastAPI 包装为 AWS Lambda 的适配器，来部署 FastAPI 应用程序：

from mangum import Mangum
from fastapi import FastAPI

app = FastAPI()

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

handler = Mangum(app)

对于更复杂的部署，AWS SAM 与 Python PowerTools 为 FastAPI 应用程序提供了出色的工具。

传统服务器部署

在传统服务器或 Kubernetes 上进行生产部署：

# 使用 Gunicorn 和 Uvicorn 工作者
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

高级功能

CORS（跨域资源共享）

为前端应用启用 CORS：

from fastapi.middleware.cors import CORSMiddleware

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

WebSocket 支持

FastAPI 支持 WebSocket 用于实时通信：

from fastapi import WebSocket

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message received: {data}")

GraphQL 集成

FastAPI 可以通过 Strawberry 或其他 GraphQL 库与 GraphQL 集成：

import strawberry
from strawberry.fastapi import GraphQLRouter

@strawberry.type
class Query:
    @strawberry.field
    def hello(self) -> str:
        return "Hello World"

schema = strawberry.Schema(query=Query)
graphql_app = GraphQLRouter(schema)

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

与 LLM 和 AI 一起工作

FastAPI 在构建 AI 驱动的 API 方面表现出色。当 使用 Ollama 对 LLM 进行结构化输出限制 时，FastAPI 的 Pydantic 模型为结构化响应提供了完美的模式定义：

from pydantic import BaseModel
from typing import List
import httpx

class SentimentResponse(BaseModel):
    sentiment: str
    confidence: float
    keywords: List[str]

@app.post("/analyze-sentiment/", response_model=SentimentResponse)
async def analyze_sentiment(text: str):
    # 调用 LLM 服务
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://localhost:11434/api/generate",
            json={
                "model": "qwen2.5",
                "prompt": f"Analyze sentiment: {text}",
                "format": SentimentResponse.schema_json()
            }
        )
    return response.json()

最佳实践和技巧

1. 一致使用路径操作装饰器

使用清晰的 HTTP 方法组织您的端点：

@app.get("/items/")      # 列出项目
@app.post("/items/")     # 创建项目
@app.get("/items/{id}")  # 获取特定项目
@app.put("/items/{id}")  # 更新项目
@app.delete("/items/{id}") # 删除项目

2. 对 API 进行版本控制

从一开始就包括 API 版本控制：

from fastapi import APIRouter

api_v1 = APIRouter(prefix="/api/v1")
api_v2 = APIRouter(prefix="/api/v2")

@api_v1.get("/users/")
async def get_users_v1():
    return {"version": "1.0"}

app.include_router(api_v1)
app.include_router(api_v2)

3. 使用环境变量

不要硬编码秘密信息。使用环境变量：

from pydantic import BaseSettings

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

settings = Settings()

4. 实现适当的日志记录

使用 Python 的日志模块：

import logging

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

@app.get("/items/")
async def get_items():
    logger.info("Fetching all items")
    return items

5. 监控性能

使用中间件进行请求计时：

import time
from fastapi import Request

@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

FastAPI 与其他框架的对比

FastAPI 与 Flask

Flask 更加极简主义，并为您提供更多控制权，但 FastAPI 提供了更多开箱即用的功能：

  • FastAPI 有自动验证和文档
  • Flask 需要扩展来支持异步功能
  • FastAPI 对 I/O 密集型操作显著更快
  • Flask 有更大的生态系统和更多的学习资源

FastAPI 与 Django REST Framework

Django REST Framework（DRF）是更大 Django 生态系统的一部分：

  • DRF 包括完整的 ORM 和管理界面
  • FastAPI 更轻量且更快
  • DRF 更适合复杂的、数据库密集型的应用程序
  • FastAPI 在微服务和独立 API 方面表现优异

FastAPI 与 Node.js/Express

FastAPI 提供了 Python 的简单性，其性能与 Node.js 相当：

  • 类似的异步性能特征
  • Python 的类型系统比 JavaScript/TypeScript 更强大
  • Node.js 有更大的包生态系统
  • FastAPI 的自动文档更优越

实际应用案例

微服务架构

FastAPI 的轻量级特性和快速启动时间使其非常适合微服务。每个服务都可以独立部署和扩展。

机器学习 API

提供 ML 模型是 FastAPI 的常见使用案例。该框架的异步支持可以高效处理多个预测请求。

移动/Web 应用的后端

FastAPI 是现代单页应用（SPA）和移动应用的绝佳后端，因为它有自动 CORS 处理和 WebSocket 支持。

物联网数据收集

FastAPI 可以高效处理来自物联网设备的高流量数据采集，处理和存储传感器数据。

聊天机器人和对话式 AI

构建对话界面和聊天机器人后端得益于 FastAPI 的 WebSocket 支持和异步能力。

常见陷阱和解决方案

1. 异步函数中的阻塞操作

问题： 在异步函数中运行同步阻塞代码：

@app.get("/slow")
async def slow_endpoint():
    time.sleep(10)  # 阻塞整个事件循环！
    return {"status": "done"}

解决方案： 使用 run_in_executor 处理阻塞操作：

from fastapi import BackgroundTasks
import asyncio

@app.get("/fast")
async def fast_endpoint():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, blocking_function)
    return {"result": result}

2. 不使用响应模型

问题： 返回原始数据库对象会暴露内部结构。

解决方案： 始终定义响应模型：

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    return db_user  # 自动通过 UserResponse 过滤

3. 不充分的错误处理

问题： 未处理的异常会崩溃应用程序。

解决方案： 使用异常处理程序：

@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
    return JSONResponse(
        status_code=400,
        content={"message": str(exc)}
    )

Python 生态系统资源

FastAPI 与更广泛的 Python 生态系统无缝协作。要全面了解 Python 基础知识，请查看 Python 快速参考，它涵盖了在 FastAPI 应用程序中使用的 Python 语法和模式。

结论

FastAPI 在 Python Web 框架设计方面取得了重大进步。通过采用现代 Python 特性如类型提示和 async/await，它提供了卓越的性能和开发体验。无论您是构建一个简单的 REST API 还是复杂的微服务架构，FastAPI 都能为您提供生产应用所需的工具和性能。

该框架的自动文档、类型安全和直观设计使其成为初学者和有经验的开发者的绝佳选择。随着 Python 生态系统的不断发展，FastAPI 将自己定位为现代 API 开发的框架。

从一个简单的 API 开始，探索文档，并随着应用程序的增长逐渐采用更高级的功能。FastAPI 社区是活跃且乐于助人的，有大量文档和示例来指导您。

有用的链接

外部资源和参考资料