WSGI与ASGI

WSGI（Web Server Gateway Interface）与ASGI（Asynchronous Server Gateway Interface）规定Python Web服务器怎样调用应用。它们不是服务器，也不是框架；它们是双方都能实现的边界契约。服务器负责HTTP解析、套接字和进程管理，框架则在这条边界内完成路由与业务处理。

WSGI面向同步HTTP。服务器为每个请求调用application(environ, start_response)，应用调用start_response()提交状态与响应头，再返回一个产生bytes的可迭代对象。PEP 3333的目标是让服务器与框架可以独立选择，而不是给业务开发者提供新的高级Web API。

ASGI把边界扩展为async application(scope, receive, send)。scope描述一条连接，receive()提供入站事件，send()提交出站事件。不同子协议定义各自的scope与消息，因此同一种应用接口可以承载HTTP、WebSocket和lifespan。

两者的差异不等于“旧而慢”和“新而快”。WSGI很适合同步框架与短HTTP请求；ASGI适合需要WebSocket、长轮询、流式请求或大量并发等待的应用。CPU工作不会因为ASGI自动提速，同步依赖也不会因为被async def包裹就停止阻塞。

你通常在部署入口、框架适配器、中间件和故障堆栈里遇到这两个名称。选择时先看应用暴露哪种callable，再看服务器、反向代理和客户端链路是否支持需要的协议，不要只看产品名称。

网络上的HTTP字节不会直接进入业务函数。服务器先把请求转换成WSGI环境或ASGI消息，应用再返回协议规定的数据。中间件同时扮演内侧应用的服务器和外侧服务器的应用，因此它必须保持两边的契约。

WSGI的一次调用

environ是内置dict，包含CGI风格字段和wsgi.input等WSGI字段。元数据与响应状态、响应头使用受规范约束的str；请求体和响应体使用字节。应用可以只读取所需字段，但不能假定所有可选字段都存在。

start_response(status, headers)设置响应状态和头部，应用返回的可迭代对象随后产生零个或多个字节块。服务器在请求正常结束、迭代报错或客户端提前断开时，都必须调用结果对象的close()（若存在）。应用也不能假定迭代器一定会被完全消费。

WSGI本身没有规定服务器要使用进程、线程还是其他调度方式。同步callable会占用执行它的worker，等待期间能否处理其他请求取决于服务器与worker模型。接口契约和并发部署是两个不同问题。

ASGI的scope与事件

ASGI服务器按连接调用应用一次。对HTTP而言，这条scope对应一个请求，即使底层HTTP/2连接复用多个流；对WebSocket而言，它持续到套接字关闭。请求体不在HTTP scope中，而是通过一个或多个http.request事件到达。

每个事件都是带有顶层type字段的dict。应用必须遵守子协议的顺序，例如先发送一次http.response.start，再发送一个或多个http.response.body；最后一个body事件把more_body设为False或省略它。WebSocket与lifespan使用不同的事件集合。

receive()和send()都是可等待callable。等待receive()让出执行权，直到有新事件；等待send()让服务器有机会把数据送入发送缓冲区并施加 背压（backpressure） 。它不表示远端客户端已经收到数据。

边界上的所有权

WSGI应用拉取wsgi.input并由服务器拉取响应迭代器；ASGI应用等待入站事件并主动发送出站事件。这个方向差异决定了中间件怎样缓存、流式转发和清理资源。把两种签名机械拼接起来不会构成正确适配器。

能力需要整条链路支持

ASGI HTTP子协议可以表示HTTP/1.0、HTTP/1.1与HTTP/2，但部署是否真正启用某个版本由服务器和上游链路决定。WebSocket也需要服务器与反向代理正确处理握手和连接升级。看到ASGI入口只能证明应用边界，不能证明生产路径已经具备所有协议能力。

WSGI应用可以通过适配器运行在ASGI服务器后面。规范要求同步WSGI callable在线程池中执行，这保留了HTTP兼容性，却不会给应用添加WebSocket或原生异步流处理。适配层还要负责字符串、字节、请求体和线程敏感资源的转换。

下面四个程序直接驱动callable，不打开端口，也不依赖第三方框架。它们依次展示WSGI往返、ASGI分片消息、lifespan和同步I/O隔离；输出来自本地python3实际运行。

最小WSGI往返

这个小型驱动器提供应用实际读取的环境字段，并捕获start_response()。finally中的close()模拟服务器即使提前结束也要履行的清理责任。

真实服务器会补齐PEP 3333要求的环境字段，并把返回字节写入网络。这个驱动器只验证应用与接口的交互，不验证HTTP解析、代理头或套接字行为。框架测试客户端与这种进程内驱动器有同样的边界限制。

处理ASGI请求分片

ASGI请求体可以跨多个http.request事件到达。应用持续读取到more_body为假，再用两个body事件生成一个逻辑响应。

import asyncio

async def application(scope, receive, send):
    assert scope["type"] == "http"
    body = bytearray()
    while True:
        message = await receive()
        if message["type"] == "http.disconnect":
            return
        body.extend(message.get("body", b""))
        if not message.get("more_body", False):
            break

    await send({"type": "http.response.start", "status": 201,
                "headers": [(b"content-type", b"text/plain")]})
    await send({"type": "http.response.body", "body": b"received=",
                "more_body": True})
    await send({"type": "http.response.body",
                "body": str(len(body)).encode(), "more_body": False})

incoming = iter([
    {"type": "http.request", "body": b"abc", "more_body": True},
    {"type": "http.request", "body": b"def", "more_body": False},
])
sent = []

async def receive():
    return next(incoming)

async def send(message):
    sent.append(message)

scope = {"type": "http", "method": "POST", "path": "/upload"}
asyncio.run(application(scope, receive, send))

for message in sent:
    if message["type"] == "http.response.start":
        print("start", message["status"], message["headers"])
    else:
        print("body", repr(message["body"]), message["more_body"])

start 201 [(b'content-type', b'text/plain')]
body b'received=' True
body b'6' False

两个输入片段合计六个字节，两个输出片段通过more_body组成一个响应。生产代码不能无限累加请求体；应在每次扩展缓冲区前后检查累计大小，超过限制就停止读取并按框架或服务器约定结束请求。

这个测试还没有模拟send()异常或任务取消。长轮询与流式响应需要分别测试“断开发生在下一次receive()”和“send()先抛出OSError”两条路径，因为并发下两者顺序不固定。

驱动lifespan

lifespan在处理请求的事件循环中完成资源初始化与关闭。示例把共享状态写入scope["state"]，并明确确认startup与shutdown。

import asyncio


async def application(scope, receive, send):
    assert scope["type"] == "lifespan"
    state = scope["state"]

    while True:
        message = await receive()
        if message["type"] == "lifespan.startup":
            state["catalog"] = "ready"
            await send({"type": "lifespan.startup.complete"})
        elif message["type"] == "lifespan.shutdown":
            state["catalog"] = "closed"
            await send({"type": "lifespan.shutdown.complete"})
            return


incoming = iter([
    {"type": "lifespan.startup"},
    {"type": "lifespan.shutdown"},
])
sent = []


async def receive():
    return next(incoming)


async def send(message):
    sent.append(message)


scope = {"type": "lifespan", "state": {}}
asyncio.run(application(scope, receive, send))

for message in sent:
    print(message["type"])
print(scope["state"]["catalog"])

lifespan.startup.complete
lifespan.shutdown.complete
closed

支持lifespan state的服务器会把这个命名空间浅复制给后续请求scope。实际资源通常是连接池或客户端对象，而不是字符串。多进程服务器会在每个进程的事件循环中执行lifespan，所以初始化代码必须允许每个worker拥有自己的资源。

隔离同步I/O

已有同步客户端暂时无法替换时，可用asyncio.to_thread()把阻塞I/O移出事件循环线程。下面只验证调用与返回值，不用时间差冒充性能基准。

import asyncio
import time


def read_legacy_record(record_id):
    time.sleep(0.01)
    return {"id": record_id, "state": "paid"}


async def heartbeat():
    await asyncio.sleep(0)
    return "loop stayed runnable"


async def main():
    record, pulse = await asyncio.gather(
        asyncio.to_thread(read_legacy_record, 7),
        heartbeat(),
    )
    print(pulse)
    print(record)


asyncio.run(main())

loop stayed runnable
{'id': 7, 'state': 'paid'}

to_thread()主要用于否则会阻塞事件循环的I/O函数。它不是把任意CPU工作变快的开关，也不会自动让底层调用响应协程取消。线程池容量、连接池容量、超时和关闭行为仍需一起设计。

把async def当成非阻塞保证

修复方法： 优先使用真正的异步库；暂时保留I/O型同步函数时，通过框架提供的同步边界或asyncio.to_thread()隔离，并测试超时、取消与线程池耗尽。CPU密集工作应使用适合它的进程或任务系统。

假定请求体已经完整

修复方法： WSGI按已验证的CONTENT_LENGTH或框架API读取；ASGI循环处理more_body，同时累计并强制大小上限。不要为了方便而把未知大小的请求完整读入内存。

把响应头转换成字典

修复方法： 把头部当作有序的二元组序列，按协议要求追加、替换或删除单个字段。为重复头、非ASCII路径以及空响应体增加边界测试，不要只测普通JSON响应。

破坏响应顺序与清理

修复方法： 用状态机测试合法事件序列，并在finally中释放由请求拥有的资源。分别覆盖正常完成、应用异常、客户端断开和服务器取消，不要把其中一种路径当作其余路径的证明。

把适配器当成能力升级

修复方法： 写清需要保留的子协议、流式行为、线程亲和性、上下文传播和取消语义。只在这个交集内使用适配器，并用真实服务器路径验证代理、超时与断开连接。

在错误的生命周期共享资源

修复方法： 在ASGI lifespan中按事件循环创建并关闭资源，通过request scope state或框架等价机制访问。明确检查单进程、多worker、重载和启动失败路径；WSGI资源则应遵守所选进程与线程模型的生命周期钩子。

• PEP 3333：Python Web Server Gateway Interface v1.0.1
• ASGI 3.0：主规范
• ASGI 3.0：HTTP与WebSocket消息格式
• ASGI 3.0：lifespan协议
• Python 3.14 asyncio：在线程中运行阻塞函数