aiohttp,一个高效的 Python 库

大家好，我是木木。

今天给大家分享一个高效的 Python 库，aiohttp。

aiohttp

如果你平时既要写 HTTP 客户端，又要落异步 API 服务，甚至还要顺手接上 WebSocket，那 aiohttp 这种客户端和服务端一体化的库会非常顺手。它把请求、路由、中间件和长连接都放在同一套 asyncio 语境里，特别适合高并发接口、内部工具服务和需要实时回传状态的场景。

项目地址：https://github.com/aio-libs/aiohttp

官方文档：https://docs.aiohttp.org

三大特点

双端统一

客户端请求和服务端路由都在同一套异步模型里，排查链路和复用代码都更自然。

中间件顺手

鉴权、日志、重试和统一响应逻辑都能挂在中间件层，业务代码更容易保持干净。

实时友好

WebSocket 开箱即用，做控制台推送、机器人通道和轻量实时协作都很合适。

最佳实践

安装方式：python -m pip install aiohttp

这篇我不去依赖外部网站，而是每段代码都用一份本地 aiohttp 应用自举测试：先在脚本里起服务，再用 ClientSession 或 WebSocket 客户端打过去。这样更接近真实开发时的最小闭环，也能把输出控制得更稳定。

功能一：同一套代码里同时管理异步服务和异步请求

这段代码解决什么问题：很多人第一次接触 aiohttp 时，只把它当作异步版的 HTTP 客户端。但它真正顺手的地方是，你可以在同一个 asyncio 流程里先起一个 Web 服务，再立刻用 ClientSession 去打这个服务，把参数、请求头和响应 JSON 全部串起来验证。

importasynciofromaiohttpimportweb,ClientSessionPORT=18081asyncdefhandle(request):returnweb.json_response({"query":request.query.get("q"),"limit":int(request.query.get("limit","0")),"trace":request.headers.get("X-Trace-Id"),})asyncdefmain():app=web.Application()app.router.add_get("/search",handle)runner=web.AppRunner(app)awaitrunner.setup()site=web.TCPSite(runner,"127.0.0.1",PORT)awaitsite.start()try:asyncwithClientSession()assession:asyncwithsession.get(f"http://127.0.0.1:{PORT}/search",params={"q":"aiohttp","limit":"2"},headers={"X-Trace-Id":"demo-01"},)asresp:print("status:",resp.status)data=awaitresp.json()print("query :",data["query"])print("limit :",data["limit"])print("trace :",data["trace"])finally:awaitrunner.cleanup()asyncio.run(main())

这种写法很适合接口调试和集成测试。你不用把服务端和客户端拆成两套心智模型，很多请求链路的问题都能在一个脚本里先快速复现出来。

功能二：鉴权和访问控制可以直接挂到中间件层

这段代码解决什么问题：真实项目里，API 通常不只是“收请求然后回 JSON”，还要先做 token 校验、拒绝匿名访问、统一错误格式。aiohttp.web 的中间件接口很轻，适合把这些横切逻辑提前收好。

importasynciofromaiohttpimportweb,ClientSessionPORT=18082@web.middlewareasyncdefauth_middleware(request,handler):ifrequest.headers.get("Authorization")!="Bearer demo-token":returnweb.json_response({"error":"missing or invalid token"},status=401)returnawaithandler(request)asyncdefdashboard(request):returnweb.json_response({"ok":True,"section":"admin","user":"wood"})asyncdefmain():app=web.Application(middlewares=[auth_middleware])app.router.add_get("/admin",dashboard)runner=web.AppRunner(app)awaitrunner.setup()site=web.TCPSite(runner,"127.0.0.1",PORT)awaitsite.start()try:asyncwithClientSession()assession:asyncwithsession.get(f"http://127.0.0.1:{PORT}/admin")asresp:print("anon  :",resp.status,awaitresp.text())asyncwithsession.get(f"http://127.0.0.1:{PORT}/admin",headers={"Authorization":"Bearer demo-token"},)asresp:print("authed:",resp.status,awaitresp.text())finally:awaitrunner.cleanup()asyncio.run(main())

这个模式在内部后台、运维接口和带登录态的控制面服务里很常见。把鉴权拦截放在中间件里，后面加日志、审计或请求 ID 也会更顺。

环境与版本信息

• Demo 环境：Windows 11，Python 3.11
• 本文安装的 aiohttp 版本：3.13.5
• 随包安装的关键依赖包括：aiohappyeyeballs、aiosignal、frozenlist、multidict、propcache、yarl
• 仓库 pyproject.toml 当前要求 Python >=3.10
• GitHub 最近一次推送时间：2026-04-22T14:51:37Z

高级功能

这段代码解决什么问题：一旦场景从普通请求升级到实时互动，很多人就会拆去别的库单独做长连接。但 aiohttp 自己就把 WebSocket 收进来了，所以你可以继续复用同样的事件循环、路由和生命周期管理。

importasynciofromaiohttpimportweb,ClientSession,WSMsgTypePORT=18083asyncdefwebsocket_handler(request):ws=web.WebSocketResponse(heartbeat=10)awaitws.prepare(request)asyncformsginws:ifmsg.typeisWSMsgType.TEXT:awaitws.send_str(f"ACK:{msg.data.upper()}")elifmsg.typeisWSMsgType.ERROR:breakreturnwsasyncdefmain():app=web.Application()app.router.add_get("/ws",websocket_handler)runner=web.AppRunner(app)awaitrunner.setup()site=web.TCPSite(runner,"127.0.0.1",PORT)awaitsite.start()try:asyncwithClientSession()assession:asyncwithsession.ws_connect(f"http://127.0.0.1:{PORT}/ws")asws:awaitws.send_str("ping wood")reply=awaitws.receive()print("ws type :",reply.type)print("ws text :",reply.data)finally:awaitrunner.cleanup()asyncio.run(main())

这类能力很适合做日志尾流、任务进度推送、机器人会话和轻量实时面板。好处是不必为了一个 WebSocket 通道再额外引入另一套服务框架。

适用场景

• 你需要同时写异步 HTTP 客户端和异步 Web 服务，希望两边共用一套 asyncio 语境
• 你的接口层需要中间件、鉴权、统一响应结构或轻量实时能力
• 你在做网关、内部工具服务、机器人后端或高并发 API

不适用场景

• 你的团队主要是同步代码栈，暂时没有准备好管理 asyncio 生命周期和并发模型
• 你需要的是重型全栈后台框架，不只是 HTTP 层和异步服务能力
• 你的场景几乎全是一次性脚本请求，不需要长期维护的异步服务或连接

上线检查

1. 先统一服务端超时、连接池和清理策略，避免请求挂起后拖住整个事件循环
2. 中间件里把鉴权、日志和异常格式先收敛好，不要把这些逻辑散在每个 handler 里
3. 如果会上 WebSocket，提前确认心跳、断连重试和消息大小限制

总结

如果你想把异步请求、异步服务和实时连接放进一套高效又统一的工作流里，aiohttp 依然是非常能打的一档选择。