【译】FastAPI 解构：现代 ASGI 框架的剖析

【由腾讯元宝辅助翻译】

文章来源：https://rafiqul.dev/blog/fastapi-deconstructed-anatomy-of-modern-asgi-framework

发布时间：2024-11-19T15:32:14.000Z

最近，我有机会在 PyCon APAC 2024 上深入探讨 FastAPI 的内部机制。演讲的标题是“FastAPI 解密：现代 ASGI 框架的剖析”。随后，我想到为什么不将这次演讲的内容写成文字呢？于是，我决定写一篇博客文章。以下就是这篇文章。

你可以在这里找到演讲的幻灯片：https://github.com/shopnilsazal/fastapi-deconstructed

注意： 为了更好地查看图表，请使用浅色模式阅读本文。

FastAPI 迅速成为 Python 开发者首选的框架之一，尤其是在需要高性能和开发者友好的 API 框架时。凭借对异步编程、依赖注入和自动生成 OpenAPI 文档的支持，FastAPI 以其速度和易用性脱颖而出。本文将分解 FastAPI 的核心组件，详细说明从 ASGI 和 Uvicorn 到 Starlette 和 Pydantic 的各个部分如何协同工作，构建出一个强大且现代的 Web 框架。

Hello World

让我们从 FastAPI 应用程序的基础开始。在 FastAPI 中，一个“Hello World”示例非常简单。

from fastapi import FastAPI

app = FastAPI()

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

通过这样一个简单的设置，FastAPI 可以处理以下任务：

定义异步路由。
解析和验证请求。
序列化 JSON 响应。
自动生成 API 文档。

我们可以通过以下方式运行这个应用程序：

1	uvicorn main:app

1	hypercorn main:app

1	granian --interface asgi main:app

可以看到，有多种方式可以运行我们的应用程序。关键在于，我们需要一个符合 ASGI 协议的服务器来运行应用程序。我们可以使用任何实现 ASGI 协议的服务器。但为了简单起见，在本文中我将使用 uvicorn 作为 ASGI 服务器的示例来解释相关内容。

构建模块

FastAPI 的功能建立在多个强大的组件之上：

ASGI：异步协议层，负责处理服务器与应用程序之间的通信。
Uvicorn：一个高性能的 ASGI 服务器，用于运行 FastAPI 应用程序。
Starlette：一个 ASGI 框架，提供路由、中间件和请求/响应处理功能。
Pydantic：用于数据验证和解析的库，FastAPI 使用它来确保数据的一致性和可靠性。
依赖注入：内置的依赖注入系统，使得注入数据库连接、服务或配置等依赖变得非常容易。
自动 API 文档：自动生成 OpenAPI 规范，提供详细的文档和交互功能。

ASGI - 协议层

ASGI（异步服务器网关接口）是 FastAPI 的基础，它通过提供应用程序与服务器之间的标准化接口来支持异步编程。ASGI 从 WSGI（Web 服务器网关接口）演变而来，以支持实时 Web 功能，如 WebSocket 和多个并发连接，使得 Python 应用程序能够处理高负载而不会阻塞。目前，ASGI 协议描述了 HTTP/1.1、HTTP/2 和 WebSocket。

从高层来看，ASGI 应用程序的请求流程如下：当客户端发送 HTTP 请求时，ASGI 服务器接受请求并将其解析并转换为 scope 和 events（稍后我们会详细讨论 scope 和 events）。然后，ASGI 应用程序接收 scope 和 events 并处理请求。现在让我们看看 ASGI 协议本身的一些细节。

ASGI 组件：

Scope：ASGI 为每个连接定义了一个 scope。这是一个包含连接元数据的字典。对于 HTTP 请求，它包括方法、路径、查询字符串、标头等。每个请求或连接都封装在一个唯一的 scope 中。

示例 HTTP scope：

scope = {
    "type": "http",  # 连接类型（"http", "websocket"）
    "http_version": "1.1",  # HTTP 版本
    "method": "GET",  # HTTP 方法，如 GET、POST
    "path": "/hello",  # 客户端请求的 URL 路径
    "query_string": b"name=John",  # 请求中的查询字符串
    "headers": [  # HTTP/Websocket 标头
        (b"host", b"example.com"),
        (b"user-agent", b"Mozilla/5.0"),
        (b"accept", b"text/html"),
    ],
    "client": ("127.0.0.1", 12345),  # 客户端 IP 地址和端口
    "server": ("127.0.0.1", 8000),  # 服务器 IP 地址和端口
}

Events：ASGI 通过事件来处理请求。事件是用于接收传入数据或发送传出数据的异步函数：
- Receive：一个 awaitable 的可调用对象，应用程序调用它来接收事件（如 HTTP 请求或 WebSocket 消息）。
- Send：一个 awaitable 的可调用对象，应用程序使用它来将响应发送回服务器。
Lifespan Events：ASGI 还支持生命周期事件，用于处理启动和关闭操作。这些事件允许在服务器启动或停止时运行设置或清理任务（如初始化或关闭数据库连接）。

这是一个简单的 ASGI 应用程序的样子。没有框架，只是一个简单的 Python 异步函数。

async def app(scope, receive, send):
    assert scope['type'] == 'http'

    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [
            [b'content-type', b'text/plain'],
        ],
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello, world!',
    })

Uvicorn - ASGI 服务器

Uvicorn 是运行 FastAPI 应用程序的 ASGI 服务器。你可以使用任何其他 ASGI 服务器来运行 FastAPI 应用程序。Uvicorn 设计用于速度和效率，使其成为需要高并发应用程序的理想选择。Uvicorn 建立在 uvloop 之上，这是一个高性能的 asyncio 事件循环实现，增强了其高效处理 I/O 密集型任务的能力。

Uvicorn 中的请求生命周期：

接受连接：Uvicorn 接受连接并为传入的 HTTP 请求创建 ASGI scope，包括标头和方法等元数据。
分发请求：scope 被分发给 FastAPI 应用程序。Uvicorn 使用 uvloop 来异步管理流程。
接收数据：Uvicorn 通过 ASGI receive 事件处理传入的请求数据。
发送响应：FastAPI 通过 ASGI send 事件响应。Uvicorn 打包响应（状态码、标头、正文）并将其返回给客户端。

Starlette - ASGI 框架层

谈到 FastAPI，我们不能不提 Starlette。Starlette 是一个轻量级的 ASGI 框架，为 FastAPI 提供了其核心功能。Starlette 作为 FastAPI 的骨干，处理底层的路由、中间件和 ASGI 兼容性，而 FastAPI 则添加了 Pydantic 验证、依赖注入和构建 API 的额外工具。

Web 请求的生命周期

现在，让我们通过一个 starlette 的 hello world 示例来可视化 HTTP 请求的完整生命周期。

Starlette 应用程序 (app.py)：

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

# 处理 "/hello" 的路由
async def hello(request):
    return JSONResponse({'message': 'Hello, World!'})

# 定义路由
routes = [
    Route('/hello', hello),
]

# 创建 Starlette 应用
app = Starlette(debug=True, routes=routes)

使用 Uvicorn 运行应用程序

1	uvicorn app:app --host 127.0.0.1 --port 8000

客户端请求：

1	curl http://127.0.0.1:8000/hello

这将返回 JSON 响应 {"message": "Hello, World!"}。

现在，让我们一步步跟踪请求，从客户端发送 HTTP 请求到返回响应的过程。

步骤 1：客户端发送 HTTP 请求

客户端向服务器发送 HTTP 请求。例如，向 /hello 端点发送 GET 请求。

GET /hello HTTP/1.1
Host: 127.0.0.1:8000
User-Agent: curl/7.64.1
Accept: */*

步骤 2：Uvicorn 接受请求

Uvicorn 运行一个套接字服务器，监听指定主机/端口（如 127.0.0.1:8000）上的传入 TCP 连接。当 HTTP 请求到达时，Uvicorn：

接受 TCP 连接。
使用 h11（一个纯 Python 的 HTTP/1.1 库）或 httptools（nodejs HTTP 解析器的 Python 绑定）从原始 TCP 数据中解析 HTTP 请求。

在这里，Uvicorn 会将传入的请求转换为 ASGI scope 和 events。

步骤 3：Uvicorn 将 HTTP 请求转换为 ASGI Scope

当 Uvicorn 收到 HTTP 请求时，它会将其转换为 ASGI scope 对象。

scope = {
    "type": "http",
    "http_version": "1.1",
    "method": "GET",
    "path": "/hello",
    "query_string": b"",
    "headers": [
        (b"host", b"127.0.0.1:8000"),
        (b"user-agent", b"curl/7.64.1"),
        (b"accept", b"*/*"),
    ],
    "client": ("127.0.0.1", 12345),
    "server": ("127.0.0.1", 8000),
}

Type：连接类型，对于 HTTP 请求是 http。
HTTP Version：HTTP 协议的版本（如 1.1）。
Method：请求中使用的 HTTP 方法（GET、POST 等）。
Path：请求的 URL 路径（如 /hello）。
Headers：标头键值对的列表。
Client：客户端的 IP 和端口。
Server：服务器的 IP 和端口。

步骤 4：Uvicorn 将 Scope 传递给 ASGI 应用程序

一旦 Uvicorn 创建了 ASGI scope，它将通过调用应用程序的可调用对象来启动 ASGI 应用程序（在本例中为 Starlette）：

1 2	async def app(scope, receive, send): ...

Uvicorn 调用 Starlette 应用程序，传入 scope 对象。

步骤 5：Starlette 处理请求

Starlette 作为一个符合 ASGI 的框架，此时接管了请求。它匹配路由（在本例中为 /hello）并调用相应的路由处理程序。

在本例中，当请求 /hello 路由时，会调用 hello 函数。Starlette 内部使用 ASGI scope 来匹配传入请求的方法和路径与定义的路由。

Request 对象：Starlette 从 Uvicorn 接收的 scope 和 ASGI 事件中创建 HTTP 请求对象。
接收事件（receive）：Starlette 接收代表 HTTP 请求部分的事件，包括请求体。

request_event = {
    "type": "http.request",
    "body": b"",  # 请求体
    "more_body": False,  # 表示是否还会发送更多数据
}

body 字段包含请求体（在 POST 请求的情况下），more_body 告诉应用程序请求体是否完整或是否会有更多数据（适用于流式传输大文件）。

响应处理：hello 路由返回一个 JSONResponse，它将响应数据包装起来并通过 ASGI 事件发送回去。

步骤 6：Starlette 返回响应

在处理完请求后，Starlette 通过发出 http.response.start 和 http.response.body 等 ASGI 事件将响应返回给 Uvicorn：

开始响应 (http.response.start)：

response_start_event = {
    "type": "http.response.start",
    "status": 200,  # HTTP 状态码
    "headers": [
        (b"content-type", b"application/json"),
    ],
}

这告诉 Uvicorn 开始发送 HTTP 响应标头，状态码为 200，内容类型为 application/json。

发送响应体 (http.response.body)：

response_body_event = {
    "type": "http.response.body",
    "body": b'{"message": "Hello, World!"}',  # JSON 响应体
    "more_body": False,
}

这发送包含 JSON 消息 {"message": "Hello, World!"} 的响应体。more_body: False 表示这是正文的最后一部分，响应已完成。

步骤 7：Uvicorn 将 HTTP 响应发送回客户端

Uvicorn 接收 Starlette 发出的 ASGI 事件并将其转换为 HTTP 响应。具体来说：

http.response.start 触发 Uvicorn 发送 HTTP 状态行和标头（如 HTTP/1.1 200 OK）。
http.response.body 将响应体（如 {"message": "Hello, World!"}）发送给客户端。

Uvicorn 在发送完所有响应部分后关闭连接。

FastAPI - 高层框架

FastAPI 建立在 Starlette 之上，创建了一个非常适合开发 RESTful API 的框架。FastAPI 专注于异步编程、使用 Pydantic 进行数据验证以及依赖注入，使其功能强大且易于使用。

主要特性：

基于 Starlette 的路由和请求处理。
基于 Pydantic 的数据验证。
依赖注入系统。
自动生成 OpenAPI 和 API 文档。

Pydantic - 数据验证和序列化

FastAPI 的数据验证依赖于 Pydantic，这是一个简化复杂数据类型处理和验证的库。Pydantic 使 FastAPI 能够对传入请求数据和传出响应数据实施严格的数据验证规则。

Pydantic 模型示例：

from fastapi import FastAPI
from pydantic import BaseModel

# 初始化 FastAPI 应用（类似于初始化 Starlette）
app = FastAPI()

# 用于请求数据验证的 Pydantic 模型
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = None

# 带有路径参数和 Pydantic 请求体验证的路由
@app.post("/items/")
async def create_item(item: Item):
    return {"item": item}

在上面的示例中：

客户端发送请求：

1	curl -X POST "<http://127.0.0.1:8000/items/>" -H "Content-Type: application/json" -d '{"name": "Table", "price": 150.0}'

FastAPI 验证请求体并返回：

{
  "item": {
    "name": "Table",
    "price": 150.0,
    "is_offer": null
  }
}

如果缺少必填字段（如 name），FastAPI 将返回自动验证错误，HTTP 状态码为 422：

{
  "detail": [
    {
      "loc": ["body", "name"],
      "msg": "field required",
      "type": "value_error.missing"
    }
  ]
}

Pydantic 还会根据需要转换数据类型，使得处理复杂数据时无需编写大量验证代码。

FastAPI 中的依赖注入

FastAPI 的依赖注入系统通过将数据库连接、认证层或共享配置等资源直接注入路由函数，实现了模块化和可重用的代码。

依赖注入示例：

from fastapi import Depends

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

@app.get("/items/")
async def read_items(db=Depends(get_db)):
    return db.fetch_all_items()

通过 Depends，FastAPI 自动管理依赖关系，使代码更加整洁、模块化且易于测试。依赖注入特别适用于管理外部服务，因为它允许集中控制资源的生命周期。

FastAPI 中的 OpenAPI 和 Swagger 文档

FastAPI 的自动文档生成功能提供了 Swagger 和 ReDoc 界面，无需额外设置。通过使用路由定义、参数类型和数据模型，FastAPI 创建了实时的 OpenAPI 文档，使得测试和集成 API 端点变得更加容易。

通过在 /docs（Swagger UI）和 /redoc（ReDoc）提供文档，FastAPI 为开发者提供了一种快速且交互式的方式来探索 API 路由，使得团队和外部开发者更容易使用 API。

FastAPI 中的请求生命周期

以下是请求在 FastAPI 中的处理流程总结：

客户端发送请求：客户端向服务器发送 HTTP 请求。
Uvicorn（ASGI 服务器）：Uvicorn 接收请求并创建 ASGI scope。
Starlette（路由）：Starlette 根据路径和方法将请求路由到正确的端点。
FastAPI 端点：FastAPI 处理任何依赖关系，使用 Pydantic 验证传入数据，并处理请求。
响应：Uvicorn 接收来自 FastAPI 的响应并将其发送回客户端。

结论

FastAPI 的架构结合了多个组件，以实现快速、可靠且易于使用的 API 框架：

ASGI 是现代 Python Web 框架的基础，支持异步操作。
Uvicorn 提供高效的连接处理。
Starlette 是核心 Web 框架，负责路由和中间件。
FastAPI 在 Starlette 的基础上，通过 Pydantic 进行数据验证、依赖注入和自动生成 API 文档。