Starlette源码深度解析：6500行代码打造FastAPI基石，ASGI洋葱模型与流式响应全拆解

发布时间：2026-05-25 12:10 | 原文链接

Starlette 1.0.0 是 Python ASGI 生态的基石框架——FastAPI、MCP、sse-starlette 全都构建于它之上。本文逐文件拆解 Starlette 源码，覆盖 ASGI 协议适配、洋葱模型中继、URL 路由编译、流式响应、后台任务等核心机制，每一段关键代码都配中文注解和 Mermaid 时序图，适合想深入理解 Python Web 底层或定制 ASGI 行为的中高级后端工程师阅读。

目录
  1. 概述
     1.1 Starlette 是什么
     1.2 ASGI 协议简述
  2. 核心架构
     2.1 Starlette 应用类
     2.2 中间件洋葱模型
     2.3 请求生命周期时序
  3. 源码分析
     3.1 路由系统（routing.py）
     3.2 中间件引擎（middleware/base.py）
     3.3 请求对象（requests.py）
     3.4 响应系统（responses.py）
     3.5 后台任务（background.py）
     3.6 数据结构（datastructures.py）
  4. 功能详解
     4.1 URL 路由匹配
     4.2 中间件开发模式
     4.3 流式响应
     4.4 WebSocket 支持
     4.5 Lifespan 生命周期
  5. 技术亮点
  6. 实践指南
  7. 总结
    参考文献
1. 概述

1.1 Starlette 是什么

Starlette 是一个轻量级 ASGI（Asynchronous Server Gateway Interface）框架，由 Django REST Framework 的作者 Tom Christie 创建。2026 年 3 月 22 日发布 1.0.0 正式版。它仅依赖 anyio 和 typing-extensions 两个第三方包，34 个 .py 文件合计约 6500 行代码。

Starlette 在 Python Web 生态中的位置类似"中间件层"——它不提供 ORM、模板引擎或表单验证，而是专注于高性能 HTTP/WebSocket 处理。FastAPI 就是在其基础上添加了 Pydantic 类型验证和自动 OpenAPI 文档生成。

  • 版本：1.0.0
  • 协议：BSD-3-Clause
  • 依赖：anyio、typing-extensions
  • 被依赖方：FastAPI、MCP（Model Context Protocol）、sse-starlette

1.2 ASGI 协议简述

ASGI 是 WSGI 的异步继任者。一个 ASGI 应用是一个可调用对象：

  • 1
  • 2
async def app(scope: dict, receive: callable, send: callable) -> None:    ...
  • scope：连接元信息（type、method、path、headers 等）
  • receive：异步可调用，返回事件（http.request、websocket.receive 等）
  • send：异步可调用，发送事件（http.response.start、http.response.body 等）

ASGI 支持三种 scope 类型：http、websocket、lifespan。Starlette 对三者都有完整实现。

2. 核心架构

2.1 Starlette 应用类

Starlette 的入口是 applications.py 中的 Starlette 类。它有两层结构：

  • 外层 Starlette：持有状态、异常处理器、用户中间件列表
  • 内层 Router：负责 URL 匹配和路由分发
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
class Starlette:    def __init__(self, debug=False, routes=None, middleware=None,   exception_handlers=None,lifespan=None):   self.state=State() # 应用级状态        self.router = Router(routes, lifespan=lifespan) # 路由引擎        self.exception_handlers = dict(exception_handlers or {})        self.user_middleware = list(middleware or [])   self.middleware_stack=None # 延迟构建

关键设计：middleware_stack 在首次请求时才构建（懒初始化），避免导入期开销：

  • 1
  • 2
  • 3
  • 4
  • 5
async def __call__(self, scope, receive, send):    scope["app"] = self    if self.middleware_stack is None:        self.middleware_stack = self.build_middleware_stack()    await self.middleware_stack(scope, receive, send)

2.2 中间件洋葱模型

build_middleware_stack 方法构建经典的洋葱模型：

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
def build_middleware_stack(self):    middleware = (        [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]        + self.user_middleware        + [Middleware(ExceptionMiddleware, handlers=exception_handlers, debug=debug)]    )    app = self.router    for cls, args, kwargs in reversed(middleware):        app = cls(app, *args, **kwargs)    return app

两层内置中间件由框架自动注入：

  • 最外层 ServerErrorMiddleware：捕获 500 异常，返回调试回溯
  • 最内层 ExceptionMiddleware：处理业务异常（HTTPException、WebSocketClose）

用户中间件夹在两者之间，形成：ServerError → UserMW1 → UserMW2 → Exception → Router

2.3 请求生命周期时序

流程执行说明：

  • 步骤 1-3：ASGI 服务器将原始 HTTP 请求转换为 ASGI 协议消息，调用 Starlette 应用
  • 步骤 4-7：请求沿中间件栈逐层向内传递。外层 ServerErrorMiddleware 捕获所有未处理异常，内层 ExceptionMiddleware 将 HTTPException 转换为标准错误响应
  • 步骤 8-10：Router 遍历路由表，依次调用每个 Route 的 matches() 方法，找到 FULL match 后调用 handle()
  • 步骤 11-14：响应沿中间件栈逐层向外返回。每层中间件都有机会在返回路径上修改响应
  • 关键设计：中间件栈用 reversed() 构建——最先注册的中间件离客户端最近，最后注册的离 Router 最近
3. 源码分析

3.1 路由系统（routing.py）

3.1.1 路由类型层次

路由系统定义了 4 种路由类型，全部继承自 BaseRoute：

Match 枚举三种状态：NONE（不匹配）、PARTIAL（路径匹配但 Method 不匹配）、FULL（完全匹配）。PARTIAL 用于区分 404 和 405 状态码：

  • 1
  • 2
  • 3
  • 4
class Match(Enum):    NONE = 0    # 路径不匹配 → 404    PARTIAL = 1 # 路径匹配但方法不匹配 → 405    FULL = 2    # 完全匹配 → 正常处理

3.1.2 URL 路径编译

compile_path 是路由系统的核心函数，它将 URL 模板转换为正则表达式。输入 /{username:str}/posts/{post_id:int}，输出三个值：

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
def compile_path(path: str) -> tuple[Pattern, str, dict[str, Convertor]]:    # 输入: "/{username:str}/posts/{post_id:int}"    # 输出:    #   regex:      "^/(?P<username>[^/]+)/posts/(?P<post_id>[0-9]+)$"    #   format:     "/{username}/posts/{post_id}"    #   convertors: {"username": StringConvertor(), "post_id": IntegerConvertor()}

内置转换器：str（[^/]+）、int（[0-9]+）、float、uuid、path（.*，匹配含斜杠的路径）。转换器在 convertors.py 中定义，通过 CONVERTOR_TYPES 字典注册。

3.1.3 请求-响应包装

request_response 函数将 func(request) -> response 视图函数转换为 ASGI 应用：

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
def request_response(func):    f = func if is_async_callable(func) else functools.partial(run_in_threadpool, func)    # 同步函数自动包装为线程池执行
    async def app(scope, receive, send):        request = Request(scope, receive, send)        async def app(scope, receive, send):            response = await f(request)            await response(scope, receive, send)        await wrap_app_handling_exceptions(app, request)(scope, receive, send)    return app

关键设计：同步视图函数通过 run_in_threadpool 在 anyio 线程池中执行，不会阻塞事件循环。异常通过 wrap_app_handling_exceptions 统一捕获，确保 HTTPException 被正确转换为 HTTP 错误响应。

3.2 中间件引擎（middleware/base.py）

3.2.1 BaseHTTPMiddleware 双流隔离

BaseHTTPMiddleware 是用户开发自定义中间件的基类，实现了流读取隔离和并发安全。核心机制：

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
class BaseHTTPMiddleware:    async def __call__(self, scope, receive, send):        request = _CachedRequest(scope, receive)  # 包装原始请求        streams = anyio.create_memory_object_stream()  # 内存管道
        async def call_next(request):            # 并发执行下游 app 并捕获其输出            ...            response = _StreamingResponse(status_code=..., content=body_stream())            return response
        async with anyio.create_task_group() as task_group:   response=awaitself.dispatch_func(request, call_next)            await response(scope, wrapped_receive, send)

核心流程：

流程执行说明：

  • _CachedRequest 代理了原始 ASGI receive，允许 dispatch 函数读取请求体而不破坏下游的读取能力
  • MemoryObjectStream 创建了一个内存管道：下游 app 的 send 写入 send_stream，call_next 从 recv_stream 读取
  • TaskGroup 实现并发执行：response_sent 事件和下游 app 的输出竞速，确保在响应发送后立即返回 disconnect
  • collapse_excgroups() 将 Python 3.11+ 的 ExceptionGroup 扁平化为单一异常

3.3 请求对象（requests.py）

Request 继承自 HTTPConnection，后者继承自 Mapping[str, Any]，所以请求对象可以像字典一样访问 scope 字段：

  • 1
  • 2
  • 3
  • 4
  • 5
class HTTPConnection(Mapping[str, Any], Generic[StateT]):    # scope 字段可像字典访问: request["method"], request["path"]    def __getitem__(self, key): return self.scope[key]    def __iter__(self): return iter(self.scope)    def __len__(self): return len(self.scope)

核心设计：请求体的延迟读取。body() 在首次调用时读取完整请求体并缓存：

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
async def body(self) -> bytes:    if not hasattr(self, "_body"):        chunks = []        async for chunk in self.stream():            chunks.append(chunk)        self._body = b"".join(chunks)    return self._body

stream() 返回异步生成器，逐块 yield ASGI 消息中的 body 字段，适用于大文件上传场景。form() 方法自动检测 Content-Type 并调用对应的解析器（URL-encoded 或 multipart），支持文件上传。

3.4 响应系统（responses.py）

Starlette 1.0 的 Response 类提供了丰富的子类层次：

响应对象本身也是 ASGI 应用——它们的 __call__ 方法发送 ASGI 消息：

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
async def __call__(self, scope, receive, send):    await send({        "type": "http.response.start",        "status": self.status_code,        "headers": self.raw_headers,  # [(b"content-type", b"text/html"), ...]    })    await send({        "type": "http.response.body",        "body": self.body,        "more_body": False,    })    if self.background:        await self.background()  # 响应发送后才执行后台任务

FileResponse 使用 anyio.to_thread.run_sync 在线程池中进行文件 I/O，支持 Range 请求（断点续传）和 ETag 缓存。

3.5 后台任务（background.py）

Starlette 的 BackgroundTask 和 BackgroundTasks 实现了"响应发送后执行"模式：

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
class BackgroundTask:    def __init__(self, func, *args, **kwargs):        self.func = func        self.is_async = is_async_callable(func)
    async def __call__(self):        if self.is_async:            await self.func(*self.args, **self.kwargs)        else:            await run_in_threadpool(self.func, *self.args, **self.kwargs)

BackgroundTasks 是任务列表容器，按添加顺序串行执行。关键特性：同步函数自动在线程池运行，不会阻塞异步事件循环。常见用法包括发送邮件、写入日志、清理临时文件。

3.6 数据结构（datastructures.py）

Starlette 的数据结构设计追求不可变性（Immutable）和惰性计算（Lazy Evaluation）：

  • URL：封装 URL 解析和构造，可从 scope 或字符串构建
  • Headers：不可变 HTTP 头，大小写不敏感
  • MutableHeaders：可变版本，用于构造响应头
  • QueryParams：URL 查询参数，惰性解析（首次访问时才 parse）
  • URLPath：路径对象，支持 url_path_for() 反向查找
  • State：线程安全的可变状态存储，用于在请求生命周期内传递数据
  • FormData：multipart 表单数据，支持文件和普通字段的混合访问

Headers 的不可变设计避免了中间件间意外修改请求头。QueryParams 惰性解析避免了无查询参数请求的不必要计算。

4. 功能详解

4.1 URL 路由匹配

Router.app 方法的匹配流程：

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
async def app(self, scope, receive, send):    partial = None    for route in self.routes:        match, child_scope = route.matches(scope)   ifmatch==Match.FULL: # 完全匹配 → 立即处理            await route.handle(scope, receive, send)            return        elif match == Match.PARTIAL:  # 记录第一个部分匹配   partial=route # 用于 405 响应            partial_scope = child_scope
   ifpartialisnotNone: # 无完全匹配但有部分匹配        await partial.handle(...)     # 返回 405 Method Not Allowed        return
    # redirect_slashes: 自动处理尾部斜杠重定向    if self.redirect_slashes and route_path != "/":        # 尝试添加/移除尾部斜杠后重新匹配
    await self.default(scope, receive, send)  # 404 Not Found

路由按注册顺序匹配，首个 FULL match 胜出。此设计使路由顺序敏感——更具体的路由应注册在更通用的之前。

4.2 中间件开发模式

自定义中间件只需继承 BaseHTTPMiddleware 并实现 dispatch 方法：

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
class TimingMiddleware(BaseHTTPMiddleware):    async def dispatch(self, request: Request, call_next):        start = time.monotonic()        response = await call_next(request)  # 调用下游        elapsed = time.monotonic() - start        response.headers["X-Process-Time"] = str(elapsed)        return response

call_next(request) 是分界点——之前的代码在请求路径上执行，之后的代码在响应路径上执行。多个中间件层层包裹形成洋葱模型。

4.3 流式响应

StreamingResponse 支持异步生成器作为内容源：

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
async def slow_numbers():    for i in range(10):        yield f"data: {i}\n\n".encode()        await anyio.sleep(1)
async def sse_endpoint(request):    return StreamingResponse(        slow_numbers(),        media_type="text/event-stream"    )

底层通过 ASGI 的 more_body: True 消息实现分块传输。生成器逐块 yield 内容，客户端逐块接收，无需等待完整响应。

4.4 WebSocket 支持

WebSocket 端点使用 WebSocketRoute，其匹配和握手逻辑与 HTTP 路由一致，但 scope.type 为 "websocket"。WebSocket 类封装了 ASGI WebSocket 协议：

  • ws.accept() — 接受连接（发送 websocket.accept）
  • ws.receive_text() / ws.receive_json() / ws.receive_bytes() — 接收消息
  • ws.send_text() / ws.send_json() / ws.send_bytes() — 发送消息
  • ws.close(code) — 关闭连接

无路由匹配时，Router 通过 WebSocketClose 发送 websocket.close（code=1000），客户端收到正常关闭信号而非连接重置。

4.5 Lifespan 生命周期

Router 内置完整的 ASGI Lifespan 协议支持：

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
async def lifespan(self, scope, receive, send):    started = False    app = scope.get("app")    await receive()  # 等待 lifespan.startup    try:        async with self.lifespan_context(app) as maybe_state:   ifmaybe_stateisnotNone:                scope["state"].update(maybe_state)            await send({"type": "lifespan.startup.complete"})            started = True            await receive()  # 等待 lifespan.shutdown    except BaseException:        exc_text = traceback.format_exc()        # 发送 startup.failed 或 shutdown.failed        raise    await send({"type": "lifespan.shutdown.complete"})

支持三种 lifespan 形式：@asynccontextmanager（推荐）、async generator（已弃用）、普通 generator（已弃用）。Starlette 1.0 通过 _wrap_gen_lifespan_context 将旧式生成器自动桥接到 async context manager。

5. 技术亮点
  • 零侵入的 ASGI 抽象：Starlette 的 Request、Response、WebSocket 类完全封装了 ASGI 协议的细节，用户只需处理 Python 对象，无需关心原始 scope/receive/send 消息
  • 同步/异步透明的线程池桥接：所有接受视图函数的地方都通过 is_async_callable() 检测，同步函数自动在 anyio 线程池执行——同一个框架同时服务 Django 风格的同步视图和原生 async/await
  • 洋葱中间件的并发安全：BaseHTTPMiddleware 使用 anyio 的 MemoryObjectStream 和 TaskGroup 实现下游 app 的并发执行和流隔离，解决了"中间件读取请求体后下游读不到"的经典问题
  • 路由匹配的三态分治：Match.NONE/PARTIAL/FULL 三态设计将 404 和 405 的正确区分集成到路由引擎中，而非留给应用层处理
  • 响应本身是 ASGI 应用：Response.__call__ 直接发送 ASGI 消息，这意味着 Response 可以作为独立的 ASGI 应用使用——response(scope, receive, send) 等同于一个完整的 HTTP 处理
  • 惰性计算贯穿始终：QueryParams 惰性解析、middleware_stack 懒构建、请求体惰性读取——只有被使用的才被计算，最大化资源效率
  • 1.0 之前已有百万用户：FastAPI 的流行使 Starlette 在无声中成为 Python 异步 Web 的事实标准，1.0 版本进一步收紧了类型注解、清理了弃用 API
6. 实践指南
  • 直接使用 Starlette 构建服务（不依赖 FastAPI）适用于微服务、WebSocket 代理、轻量 API 网关等场景——不需要 ORM 或自动文档时，Starlette 本身足够
  • 中间件的 dispatch 中如果读取了 request.body()，下游 middleware 和视图函数就只能拿到 _CachedRequest 的缓存版本；如果需要原始流，用 request.stream()
  • 路由顺序影响匹配结果——把最具体的路由放在前面，通配路由（如 /{path:path}）放在最后
  • redirect_slashes=True（默认）会自动将 /path 重定向到 /path/（或反之），这在与前端 SPA 路由配合时需要特别注意
  • Lifespan 中抛出的异常会导致 startup.failed 或 shutdown.failed ASGI 消息，而不仅仅是进程退出——在容器编排环境中可以利用这一点实现优雅降级
  • BackgroundTask 在响应发送后才执行——如果需要任务的结果来构建响应，应该用 anyio.create_task_group() 而非 BackgroundTask
7. 总结

Starlette 1.0.0 用约 6500 行 Python 代码实现了一个完整的 ASGI Web 框架，其"做一件事并做好"的哲学使得它成为 FastAPI、MCP 等上层框架的可靠基石。

关键收获：

  • ASGI 三层模型（http/websocket/lifespan）是理解 Starlette 架构的钥匙——Router.app 方法中的 if/elif 分支直接映射到这三种 scope 类型
  • 洋葱中间件 + MemoryObjectStream 的组合是异步 Python 中实现可组合请求/响应处理的优雅范式，值得在自己的项目中也采用类似模式
  • 同步/异步透明的线程池桥接（is_async_callable + run_in_threadpool）使得 Starlette 可以在不牺牲异步性能的前提下兼容庞大的同步代码生态
  • 适合直接使用 Starlette 的场景：WebSocket 密集型应用、需要极简依赖的微服务、需要完全控制中间件栈的 API 网关
  • 不适合直接使用的场景：需要 ORM、表单验证、OpenAPI 文档——这些场景用 FastAPI（Starlette 的超集）更合适
  • Starlette 的源码是最佳 ASGI 学习材料——它的代码量小、注释清晰、类型完备，比阅读 uvicorn 或 daphne 的源码更容易入门异步服务器编程
参考文献

[1] Starlette GitHub 仓库：https://github.com/Kludex/starlette

[2] Starlette 官方文档：https://www.starlette.io

[3] ASGI 规范：https://asgi.readthedocs.io

[4] anyio 文档：https://anyio.readthedocs.io

[5] FastAPI（Starlette 的上层框架）：https://github.com/fastapi/fastapi

[6] Uvicorn ASGI 服务器：https://www.uvicorn.org

[7] Starlette 1.0 发布说明：https://github.com/Kludex/starlette/releases/tag/1.0.0

举报/反馈
分享到： 微博 QQ 空间
对本文内容有合作意向？
我们将在 1 个工作日内与您联系
留言咨询