FastAPI：高效构建API的完整教程 – wiki大全

作者： /

FastAPI：高效构建API的完整教程

在当今快速发展的软件世界中，构建高性能、易于维护的API是许多应用程序成功的关键。Python 生态系统为此提供了众多选择，而 FastAPI 以其卓越的性能、直观的开发体验和现代化的特性，迅速成为了 API 开发领域的明星框架。

本教程将全面深入地介绍 FastAPI，从基本概念到高级特性，再到最佳实践，帮助您高效地构建健壮、可扩展的 API。

引言：为何选择 FastAPI？

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 Python Web 框架，基于标准的 Python 类型提示。它的核心优势在于：

  1. 极速性能：FastAPI 的性能可与 Node.js 和 Go 相媲美，是目前最快的 Python Web 框架之一。这得益于其对异步编程的深度支持 (基于 Starlette) 和高效的数据验证 (基于 Pydantic)。
  2. 开发效率高：通过自动化 API 文档、强大的类型提示和依赖注入系统，FastAPI 大幅减少了重复代码，提高了开发速度。
  3. 代码质量高：强制使用类型提示，使得代码更易读、易维护，并且在开发过程中就能捕获许多潜在错误。
  4. 现代特性：内置对异步操作、WebSocket、GraphQL 等现代 Web 功能的支持，使其成为构建高性能微服务的理想选择。

FastAPI 核心优势：为何它如此高效？

FastAPI 能够实现高效的 API 构建，主要得益于以下核心特性：

  • 极速性能 (Blazing Fast Performance): 利用 Starlette (一个轻量级的 ASGI 框架) 和 Pydantic (用于数据验证和序列化)，FastAPI 能够处理高并发请求，并且执行速度非常快。它充分利用了 Python 的 async/await 语法，实现了真正的异步非阻塞 I/O。
  • 自动生成API文档 (Automatic API Documentation): 无需额外配置，FastAPI 会根据您的代码自动生成交互式的 API 文档。这包括：
    • Swagger UI: 提供了一个美观的 Web 界面，可以直接在浏览器中测试 API。
    • ReDoc: 提供另一种风格的 API 文档，专注于可读性和清晰度。
      这大大节省了手动编写和维护文档的时间，并促进了前端与后端开发人员之间的协作。
  • 类型安全 (Type Safety): FastAPI 充分利用了 Python 的类型提示 (Type Hints)。这意味着您可以使用标准的 Python 语法来声明变量、函数参数和返回值的类型。FastAPI 会在运行时利用这些类型信息进行数据验证、序列化和自动文档生成，减少了运行时错误。
  • 异步支持 (Asynchronous Support): FastAPI 是一个 ASGI 框架，原生支持 async 和 await 关键字。这使得它能够高效地处理 I/O 密集型任务（如数据库查询、外部 API 调用），而不会阻塞事件循环，从而实现更高的吞吐量。
  • 数据验证 (Data Validation with Pydantic): FastAPI 使用 Pydantic 来进行数据验证、解析和序列化。您只需用 Python 类来定义数据模型，Pydantic 就会自动验证传入请求数据的格式、类型和约束，并将其转换为相应的 Python 对象。这极大地简化了数据处理的逻辑，并确保了数据的完整性。
  • 依赖注入 (Dependency Injection): FastAPI 提供了一个强大而灵活的依赖注入系统。您可以定义“依赖项”（函数或类），FastAPI 会自动检测这些依赖项并在路径操作函数被调用时注入它们。这有助于管理数据库会话、身份验证、授权等，使代码更模块化、可测试和可重用。

入门：第一个 FastAPI 应用

让我们从一个简单的“Hello World”应用开始。

1. 安装

首先，确保您的 Python 环境已准备就绪。推荐使用 Python 3.7+。
安装 FastAPI 和 Uvicorn (一个 ASGI 服务器，用于运行 FastAPI 应用)：

bash
pip install fastapi "uvicorn[standard]"

2. 基本应用结构 (main.py)

创建一个名为 main.py 的文件，内容如下：

“`python

main.py

from fastapi import FastAPI

app = FastAPI()

@app.get(“/”)
async def read_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}
“`

代码解释：

  • from fastapi import FastAPI: 导入 FastAPI 类。
  • app = FastAPI(): 创建 FastAPI 应用程序实例。
  • @app.get("/"): 这是一个装饰器，将 read_root 函数注册为处理 HTTP GET 请求的根路径 (/) 的处理器。
  • async def read_root():: 定义一个异步函数。FastAPI 推荐使用 async def 来处理 I/O 密集型操作。
  • return {"message": "Hello World"}: 返回一个 Python 字典。FastAPI 会自动将其转换为 JSON 响应。
  • @app.get("/items/{item_id}"): 定义另一个 GET 请求处理函数，包含一个路径参数 item_id。
  • item_id: int: 类型提示 int 告诉 FastAPI item_id 应该是一个整数。FastAPI 会自动验证类型并提供友好的错误信息。
  • q: str = None: 这是一个查询参数。str = None 表示 q 是一个可选的字符串参数，默认值为 None。

3. 运行应用

在命令行中，导航到 main.py 所在的目录，然后运行：

bash
uvicorn main:app --reload

命令解释：

  • uvicorn: 调用 Uvicorn 服务器。
  • main:app: main 指的是 main.py 文件，app 指的是文件中创建的 FastAPI() 实例的变量名。
  • --reload: (可选) 启用热重载。当文件发生变化时，服务器会自动重启，这在开发过程中非常有用。

现在，打开您的浏览器，访问 http://127.0.0.1:8000，您将看到 {"message": "Hello World"}。
访问 http://127.0.0.1:8000/items/5?q=somequery，您将看到 {"item_id": 5, "q": "somequery"}。

更重要的是，访问 http://127.0.0.1:8000/docs 和 http://127.0.0.1:8000/redoc，您会发现 FastAPI 已经为您自动生成了完整的交互式 API 文档！

核心功能详解与实践

FastAPI 的强大之处在于其对 HTTP 请求的全面支持和对数据处理的优雅方式。

1. 路径操作 (Path Operations)

FastAPI 提供了针对所有 HTTP 方法的装饰器：

  • @app.get(): 处理 HTTP GET 请求 (获取资源)
  • @app.post(): 处理 HTTP POST 请求 (创建资源)
  • @app.put(): 处理 HTTP PUT 请求 (更新资源)
  • @app.delete(): 处理 HTTP DELETE 请求 (删除资源)
  • @app.patch(): 处理 HTTP PATCH 请求 (部分更新资源)
  • @app.options(), @app.head(), @app.trace() 等

示例：

“`python

main.py (在现有代码基础上添加)

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

app = FastAPI()

… (前面的 read_root 和 read_item 函数)

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

@app.post(“/items/”)
async def create_item(item: Item):
“””
创建一个新商品
“””
item_dict = item.dict()
if item.tax:
price_with_tax = item.price + item.tax
item_dict.update({“price_with_tax”: price_with_tax})
return item_dict

@app.put(“/items/{item_id}”)
async def update_item(item_id: int, item: Item):
“””
更新现有商品
“””
return {“item_id”: item_id, **item.dict()}

@app.delete(“/items/{item_id}”)
async def delete_item(item_id: int):
“””
删除商品
“””
return {“message”: f”Item {item_id} deleted successfully”}
“`

2. 请求体 (Request Body)

对于 POST、PUT 等方法，客户端通常会发送包含数据的请求体。FastAPI 使用 Pydantic 的 BaseModel 来定义请求体的结构和验证规则。

  • 定义模型: 如上例中的 Item 类，继承自 pydantic.BaseModel。
    • name: str: 声明 name 字段为必需的字符串。
    • description: Optional[str] = None: 声明 description 为可选的字符串，默认值为 None。
    • price: float: 声明 price 字段为必需的浮点数。
  • 在路径操作中使用: item: Item 作为函数参数，FastAPI 会自动：
    • 读取传入的 JSON 请求体。
    • 将其转换为 Item 类的实例。
    • 验证所有字段的类型和约束。
    • 如果验证失败，自动返回一个清晰的 HTTP 422 (Unprocessable Entity) 错误响应。

3. 查询参数 (Query Parameters)

查询参数是 URL 中 ? 之后的部分，用于过滤、分页等。

  • 可选参数: q: str = None (如 read_item 示例)，默认值 None 使其成为可选。
  • 必需参数: 如果没有默认值，FastAPI 会将其视为必需的查询参数。
    python
    @app.get("/users/")
    async def get_users(limit: int, offset: int = 0): # limit 是必需的，offset 是可选的
    return {"limit": limit, "offset": offset}
  • 数据验证与转换: FastAPI 同样会验证查询参数的类型并进行转换。例如，limit: int 会确保传入的是一个整数。

4. 路径参数 (Path Parameters)

路径参数是 URL 路径的一部分，用于标识特定的资源。

  • 定义: {item_id} 语法 (如 update_item 示例)。
  • 类型提示: item_id: int 强制 item_id 为整数。如果客户端传入非整数值，FastAPI 会自动返回 HTTP 422 错误。

5. 数据验证 (Data Validation) – Pydantic 的力量

Pydantic 不仅提供基本类型验证，还支持更复杂的验证规则和数据结构：

  • 数据类型: str, int, float, bool, list, dict, datetime, UUID 等。
  • 嵌套模型: 一个 BaseModel 可以包含另一个 BaseModel 实例，实现复杂的数据结构。
  • 字段约束:
    • Field(..., gt=0, le=100): 例如，数字必须大于 0 且小于等于 100。
    • Field(..., min_length=3, max_length=50): 例如，字符串长度在 3 到 50 之间。
    • Field(..., regex="^abc"): 正则表达式匹配。
  • 默认值和工厂函数: Field(default_factory=datetime.utcnow)。

示例：更复杂的 Item 模型

“`python
from typing import List
from pydantic import Field, HttpUrl

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

class ItemWithDetails(BaseModel):
name: str = Field(…, min_length=3, max_length=50, description=”商品的名称”)
description: Optional[str] = Field(None, max_length=300, example=”一个非常酷的小部件”)
price: float = Field(…, gt=0, description=”商品价格，必须大于0″)
tax: Optional[float] = Field(None, ge=0, le=10, description=”税费，0到10之间”)
tags: List[str] = Field(default_factory=list, description=”商品的标签列表”)
images: Optional[List[Image]] = None

@app.post(“/detailed_items/”)
async def create_detailed_item(item: ItemWithDetails):
return item
``
通过/docs` 接口查看，FastAPI 会将这些 Pydantic 约束和描述自动反映在文档中。

6. 错误处理 (Error Handling)

FastAPI 提供了 HTTPException 类来抛出标准的 HTTP 错误响应。

“`python
from fastapi import HTTPException, status

@app.get(“/items/{item_id}”)
async def read_item(item_id: int):
if item_id == 404:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=”Item not found”,
headers={“X-Error”: “There goes my item”}, # 可以添加自定义响应头
)
return {“item_id”: item_id}
“`

  • status_code: HTTP 状态码 (例如 status.HTTP_404_NOT_FOUND)。
  • detail: 错误信息，将作为 JSON 响应体中的 detail 字段返回。
  • headers: 可选的自定义响应头。

FastAPI 还允许您自定义异常处理器，以更灵活地处理特定类型的错误。

高效构建 API 的最佳实践

仅仅了解 FastAPI 的功能是不够的，遵循最佳实践才能真正构建出高效、可维护和可扩展的 API。

  1. 理解异步/等待 (Async/Await) 的时机:

    • async def 用于 I/O 密集型操作: 当您的代码需要等待外部资源（如数据库查询、文件读写、调用第三方 API）时，使用 async def。这允许 FastAPI 在等待期间切换到处理其他请求，从而提高并发性能。
    • def 用于 CPU 密集型操作: 如果您的函数执行大量计算，没有 I/O 等待，使用普通的 def。FastAPI 会在一个单独的线程池中运行这些同步函数，避免阻塞主事件循环。
    • 避免在 async def 中执行长时间阻塞的同步代码：这会阻塞整个事件循环，抵消异步的优势。如果必须，可以考虑使用 run_in_threadpool。

  2. 模块化项目结构 (Modular Project Structure):
    随着项目增长，将所有代码放在一个 main.py 文件中会变得难以管理。使用 APIRouter 来组织您的 API 模块。

    my_project/
    ├── main.py
    ├── routers/
    │ ├── __init__.py
    │ ├── items.py
    │ └── users.py
    └── dependencies.py
    └── database.py

    routers/items.py:
    “`python
    from fastapi import APIRouter

    router = APIRouter(prefix=”/items”, tags=[“items”])

    @router.get(“/”)
    async def read_items():
    return [{“name”: “Foo”}, {“name”: “Bar”}]

    @router.post(“/”)
    async def create_item():
    return {“message”: “Item created”}
    “`

    main.py:
    “`python
    from fastapi import FastAPI
    from .routers import items, users # 假设有 users.py

    app = FastAPI()

    app.include_router(items.router)

    app.include_router(users.router)

    “`
    这样可以清晰地分离不同功能模块的代码。

  3. 充分利用 Pydantic (Extensive Use of Pydantic):

    • 请求模型: 如前所述，定义请求体的结构和验证。
    • 响应模型: 使用 response_model 参数在路径操作装饰器中声明响应的结构。FastAPI 会自动过滤掉模型中未定义的字段，确保只有预期的字段返回给客户端，这对于数据安全和API契约非常重要。
      “`python
      from fastapi import FastAPI
      from pydantic import BaseModel

    class UserInDB(BaseModel):
    username: str
    hashed_password: str
    email: str
    full_name: str = None

    class UserPublic(BaseModel):
    username: str
    email: str
    full_name: str = None

    app = FastAPI()

    @app.post(“/users/”, response_model=UserPublic)
    async def create_user(user: UserInDB): # 接收完整用户数据
    # 假设这里保存用户到数据库
    return user # 但只返回 UserPublic 定义的字段
    ``
    * **配置管理**: Pydantic 的    BaseSettings` 类可以方便地从环境变量加载配置。

  4. 依赖注入管理逻辑 (Dependency Injection for Logic):

    • 数据库会话: 将数据库会话作为依赖项注入，确保每个请求都有独立的会话，并在请求结束时正确关闭。
    • 身份验证/授权: 定义依赖项来验证用户令牌并获取当前用户。
    • 通用工具函数: 将常用的业务逻辑封装为依赖项，提高代码复用性。

    “`python
    from fastapi import Depends, HTTPException, status
    from typing import Annotated

    async def get_current_user():
    # 这里模拟从 token 获取用户
    user = {“username”: “testuser”}
    if not user:
    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=”Not authenticated”)
    return user

    @app.get(“/me/”)
    async def read_me(current_user: Annotated[dict, Depends(get_current_user)]):
    return current_user
    “`

  5. 后台任务处理 (Background Tasks):
    对于不需要立即响应客户端的耗时操作（如发送邮件、生成报告、数据清理），使用 BackgroundTasks。

    “`python
    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, message=”some notification”)
    return {“message”: “Notification sent in the background”}
    “`

  6. 数据库连接管理 (Database Connection Management):

    • 连接池: 对于关系型数据库，使用连接池来管理数据库连接，避免每次请求都创建和关闭连接。
    • 生命周期事件 (Lifespan Events): 利用 FastAPI 的 lifespan 事件来在应用启动时初始化数据库连接池，并在应用关闭时清理资源。

    “`python
    from contextlib import asynccontextmanager
    from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
    from sqlalchemy.orm import sessionmaker

    DATABASE_URL = “sqlite+aiosqlite:///./test.db”
    engine = create_async_engine(DATABASE_URL)
    AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    @asynccontextmanager
    async def lifespan(app: FastAPI):
    # 启动时执行
    print(“Application startup: Initializing database connection pool”)
    yield
    # 关闭时执行
    print(“Application shutdown: Closing database connection pool”)
    await engine.dispose()

    app = FastAPI(lifespan=lifespan)

    async def get_db():
    async with AsyncSessionLocal() as session:
    try:
    yield session
    finally:
    await session.close()

    @app.get(“/items/”)
    async def get_items(db: AsyncSession = Depends(get_db)):
    # 使用 db 会话进行数据库操作
    pass
    “`

  7. 安全配置 (Secure Configuration):

    • 环境变量: 敏感信息（如 API 密钥、数据库凭据）不应硬编码。使用环境变量在部署时注入。
    • Pydantic BaseSettings: 方便地从环境变量、.env 文件等加载配置。

    “`python
    from pydantic_settings import BaseSettings

    class Settings(BaseSettings):
    database_url: str = “sqlite:///./test.db”
    secret_key: str
    api_version: str = “v1”

    class Config:
    env_file = ".env"

    settings = Settings()

    print(settings.secret_key) # 会从 .env 或环境变量 SECRET_KEY 中读取

    “`

  8. 结构化日志 (Structured Logging):
    使用 Python 的 logging 模块进行结构化日志记录，而不是简单的 print() 语句。结构化日志更容易被日志聚合工具解析、搜索和分析，对于生产环境的监控和调试至关重要。

  9. 生产环境安全与部署 (Production Environment Security & Deployment):

    • 隐藏文档: 在生产环境中，除非是公开 API，否则应禁用或限制访问 /docs 和 /redoc 接口，以避免泄露 API 结构。
    • 使用生产级 ASGI 服务器: 不要直接使用 uvicorn main:app。在生产环境，应使用 Gunicorn 或 Hypercorn 等，通常搭配 Nginx/Caddy 等反向代理。
    • 容器化: 将 FastAPI 应用容器化 (Docker) 是现代部署的常见实践。

  10. 测试 (Testing):

    • 为您的 API 编写单元测试和集成测试是至关重要的。
    • FastAPI 提供了 TestClient，可以方便地模拟请求并测试您的路径操作函数。

    “`python
    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/10?q=test”)
    assert response.status_code == 200
    assert response.json() == {“item_id”: 10, “q”: “test”}
    “`

总结

FastAPI 是一个功能强大、性能卓越的 Python Web 框架，专为构建高效的 API 而设计。通过充分利用 Python 的类型提示、Pydantic 的数据验证和其内置的异步支持，FastAPI 显著提高了开发效率、代码质量和运行时性能。

本教程涵盖了从入门到核心功能，再到最佳实践的全面指南。掌握这些知识，您将能够利用 FastAPI 的强大功能，快速、高效地构建出符合现代 Web 开发需求的 API。现在，开始您的 FastAPI 之旅，体验构建高性能 API 的乐趣吧！

—The article has been written as requested.

滚动至顶部