aiohttp Web Server 快速入门从第一个 Handler 到路由、表单、WebSocket 与重定向的完整实践指南【免费下载链接】aiohttpAsynchronous HTTP client/server framework for asyncio and Python项目地址: https://gitcode.com/gh_mirrors/ai/aiohttp导读本文是 aiohttp基于 asyncio 与 Python 的异步 HTTP 客户端/服务端框架服务端开发的速查手册围绕官方快速入门文档 docs/web_quickstart.rst 展开。你将学会如何编写请求处理器Handler并用web.run_app启动一个服务、三种等价的路由注册方式、变量路径与反向 URL 构造、JSON 响应、表单与文件上传含流式大文件处理、WebSocket 以及重定向。文中所有代码均可直接复制运行并穿插了 aiohttp/web.py、aiohttp/web_routedef.py、aiohttp/web_urldispatcher.py 等源码级实现剖析帮助你不仅会用还知道底层是怎么工作的。快速启动一个 Web 服务器要搭建一个 Web 服务器第一步是编写一个请求处理器request handler。处理器必须是一个协程它只接收一个Request实例作为参数并返回一个Response实例from aiohttp import web async def hello(request): return web.Response(textHello, world)接着创建一个Application实例并把处理器注册到某个HTTP 方法与路径组合上app web.Application() app.add_routes([web.get(/, hello)])最后调用run_app启动应用web.run_app(app)完成。打开浏览器访问http://localhost:8080/即可看到结果。完整的可运行示例见 examples/web_srv.py。如果你更喜欢路由装饰器风格可以创建一张路由表route table并注册 web-handlerroutes web.RouteTableDef() routes.get(/) async def hello(request): return web.Response(textHello, world) app web.Application() app.add_routes(routes) web.run_app(app)两种写法做的事情本质相同区别只在于口味你是偏好 Django 式的urls.py集中路由表还是 Flask 式闪亮的装饰器。aiohttp 官方文档在代码片段中交替使用这两种方式以强调它们的等价性从一种风格切换到另一种非常容易。对应的装饰器风格完整示例见 examples/web_srv_route_deco.py。run_app 做了什么启动参数源码解析web.run_app(app)是本地开发最常用的启动入口它的完整签名位于 aiohttp/web.py。除了app之外还有一组可选的启动参数参数默认值说明hostNoneTCP/IP 主机名可传字符串或可迭代的多个主机名多 host 支持portNoneTCP/IP 端口默认 8080pathNoneUnix 文件系统路径Unix socket可与 hostname 组合sockNone直接传入已创建的 socket 对象或可迭代的多个ssl_contextNonessl.SSLContext启用 HTTPSbacklog128传给TCPSite的连接队列长度reuse_address/reuse_portNonesocket 地址/端口复用选项shutdown_timeout60.0关闭应用时的超时秒数keepalive_timeout75.0keep-alive 连接超时秒数access_log/access_log_class/access_log_formataccess_logger/AccessLogger/AccessLogger.LOG_FORMAT访问日志配置debugFalse是否以调试模式运行事件循环handle_signalsTrue是否接管 SIGINT/SIGTERM 信号handler_cancellationFalse客户端断开时是否取消正在运行的 handlerloopNone显式指定事件循环内部实现上run_app会创建事件循环并启动_run_app协程见 aiohttp/web.py它根据host/path/sock参数分别创建TCPSite、UnixSite或SockSite站点对象并启动打印出 Running on ... 横幅然后以 1 小时为间隔循环asyncio.sleep(3600)保持进程存活收到KeyboardInterrupt或GracefulExit后会在finally块中调用runner.cleanup()完成优雅关闭shutdown 与 cleanup 信号、后台任务清理的完整生命周期见 docs/web_lowlevel.rst。更复杂的场景——例如需要异步管理应用生命周期、或同时服务多个 host——可以参考 docs/web_advanced.rst 中关于application runnersAppRunnerTCPSite的讲解。命令行接口CLIaiohttp.web内置了一个基础的 CLI用于在开发环境下通过 TCP/IP 快速托管一个Application$ python -m aiohttp.web -H localhost -P 8080 package.module:init_func其中package.module:init_func是一个可导入的callable它接收所有未被解析的命令行参数列表完成应用配置后返回一个Application实例def init_func(argv): app web.Application() app.router.add_get(/, index_handler) return appCLI 的解析逻辑在 aiohttp/web.py 的main()函数中-H/--hostname默认localhost-P/--port默认8080另有-U/--path指定 Unix 文件系统路径可与 hostname 组合实现同时监听 Unix 与 TCP但在不支持AF_UNIX的平台会直接报错。入口函数必须以module:function语法给出支持相对模块名不支持以.开头的相对导入解析失败会通过arg_parser.error明确提示。开发环境也可以考虑使用 aiohttp-devtools 这类第三方工具获得热重载体验。Handler一切请求处理的起点请求处理器必须是接收且仅接收一个Request参数、返回StreamResponse派生类例如Response实例的协程async def handler(request): return web.Response()处理器通过Application.add_routes注册到某个路由HTTP 方法路径的组合上可使用get、post等辅助函数app.add_routes([web.get(/, handler), web.post(/post, post_handler), web.put(/put, put_handler)])也可以使用路由装饰器routes web.RouteTableDef() routes.get(/) async def get_handler(request): ... routes.post(/post) async def post_handler(request): ... routes.put(/put) async def put_handler(request): ... app.add_routes(routes)通配 HTTP 方法与 allow_head通配的HTTP 方法由route或RouteTableDef.route支持可以让一个处理器服务于路径上任意HTTP 方法的请求app.add_routes([web.route(*, /path, all_handler)])在处理器内部可以通过BaseRequest.method属性查询请求实际使用的 HTTP 方法。另一个值得注意的行为默认情况下用GET方法注册的端点也会接受HEAD请求并返回与GET请求相同的响应头。如果想在某个路由上拒绝HEAD请求可以显式关闭web.get(/, handler, allow_headFalse)此时handler不会被HEAD请求调用服务器将返回405: Method Not Allowed。从源码看web.get的allow_head参数默认值为True见 aiohttp/web_routedef.py而UrlDispatcher.add_get在allow_headTrue时会先额外注册一条HEAD路由再注册GET路由见 aiohttp/web_urldispatcher.py。Resources 与 Routes路由表背后的架构理解 aiohttp 的路由体系需要分清三个层级Router路由器所有路由由Application.router一个UrlDispatcher实例提供服务Resource资源路由表中的一个条目对应请求的 URL路由表中至少包含一个资源Route路由对应某个HTTP 方法与web handler的绑定关系。因此当你添加一条路由时底层会同时创建对应的resource对象。库的实现会将同一路径的后续路由添加合并为所有 HTTP 方法只保留一个资源。看两个例子app.add_routes([web.get(/path1, get_1), web.post(/path1, post_1), web.get(/path2, get_2), web.post(/path2, post_2)]app.add_routes([web.get(/path1, get_1), web.get(/path2, get_2), web.post(/path2, post_2), web.post(/path1, post_1)]第一个是优化过的——把同一路径的GET/POST放在一起注册aiohttp 会为/path1、/path2各创建一个资源分别挂载对应方法的路由。路由解析时UrlDispatcher.resolve会从 URL 末尾向前逐段回溯path_safe每次rpartition(/)截取上一级配合基于 canonical path 前缀建立的资源索引快速定位候选资源再按注册顺序线性尝试解析见 aiohttp/web_urldispatcher.py。变量路径Variable Resources资源也可以有变量路径。例如路径/a/{name}/c可以匹配/a/b/c、/a/1/c、/a/etc/c等所有符合该模式请求。变量部分以{identifier}形式指定其中的identifier可以在请求处理器中通过Request.match_info映射查找到该部分匹配到的值routes.get(/{name}) async def variable_handler(request): return web.Response( textHello, {}.format(request.match_info[name]))默认情况下每个变量部分匹配正则[^{}/]对应源码 aiohttp/web_urldispatcher.py 中的DynamicResource.GOOD即不包含{、}、/的任意字符串。你还可以用{identifier:regex}的形式指定自定义正则web.get(r/{name:\d}, handler)从源码实现看DynamicResource会用两条正则\{[_a-zA-Z][_a-zA-Z0-9]*\}纯变量与\{[_a-zA-Z][_a-zA-Z0-9]*:.\}带正则解析路径模板把变量段编译成命名分组(?Pvar...)非变量段用re.escape转义后拼接成完整正则见 aiohttp/web_urldispatcher.py而完全不含{}的路径会走PlainResource其_match直接做字符串比较——源码注释明确指出这比正则匹配快约 10 倍见 aiohttp/web_urldispatcher.py。因此能用静态路径就不要引入变量静态路由可以享受这个性能优化。使用命名资源反向构造 URL路由可以被赋予一个nameroutes.get(/root, nameroot) async def handler(request): ...之后就可以用这个名字获取资源并构造 URL例如在请求处理器内部url request.app.router[root].url_for().with_query({a: b, c: d}) assert url URL(/root?abcd)更有意思的是为变量资源构造 URLapp.router.add_resource(r/{user}/info, nameuser-info)这种情况下可以传入路径的各部分url request.app.router[user-info].url_for(userjohn_doe) url_with_qs url.with_query(ab) assert url_with_qs /john_doe/info?ab底层上UrlDispatcher把命名资源维护在一张_named_resources字典中路由名被NAME_SPLIT_RE[.:-]切分后每一段都必须是合法的 Python 标识符不能是关键字如def名字不能重复见 aiohttp/web_urldispatcher.py。url_for则通过DynamicResource._formatter模板的format_map完成变量填充所以命名资源是一处定义、全局引用避免在模板和重定向逻辑里硬编码 URL 字符串。用类组织 Handler正如前面讨论的handler 可以是一等协程async def hello(request): return web.Response(textHello, world) app.router.add_get(/, hello)但有时把逻辑相近的 handler 归入一个 Python类会更方便。由于aiohttp.web不规定任何实现细节应用开发者完全可以自由地用类组织 handlerclass Handler: def __init__(self): pass async def handle_intro(self, request): return web.Response(textHello, world) async def handle_greeting(self, request): name request.match_info.get(name, Anonymous) txt Hello, {}.format(name) return web.Response(texttxt) handler Handler() app.add_routes([web.get(/intro, handler.handle_intro), web.get(/greet/{name}, handler.handle_greeting)])这里绑定的是类的实例方法注意要传入handler.handle_intro而不是Handler.handle_intro这样多个 handler 可以共享实例状态适合把一组相关接口收拢在一个类里。基于类的视图Class Based Viewsaiohttp.web原生支持class based views。可以继承View并定义处理 HTTP 请求的方法class MyView(web.View): async def get(self): return await get_resp(self.request) async def post(self): return await post_resp(self.request)这些方法应该是只接收self的协程并像普通 web-handler 一样返回响应对象请求对象通过View.request属性获取。实现好视图如上例的MyView后需要在应用的路由器中注册以下三种方式等价app.add_routes([web.view(/path/to, MyView)])routes.view(/path/to) class MyView(web.View): ...app.router.add_route(*, /path/to, MyView)该视图会处理/path/to的 GET 与 POST 请求而对未实现的 HTTP 方法抛出405 Method not allowed异常。机制上View._iter会根据request.method的小写形式在类上查找同名方法并调用找不到对应方法时通过HTTPMethodNotAllowed抛出 405并附上类中已实现方法的集合见 aiohttp/web_urldispatcher.py。完整的类视图示例参见 examples/web_classview.py。查看已注册的资源路由器中所有已注册资源可以用UrlDispatcher.resources方法查看for resource in app.router.resources(): print(resource)而注册时带name的那部分资源子集可以用UrlDispatcher.named_resources方法查看for name, resource in app.router.named_resources().items(): print(name, resource)resources()返回ResourcesView、named_resources()返回只读的MappingProxyType包装视图见 aiohttp/web_urldispatcher.py非常适合在调试或生成路由清单sitemap时使用。注册路由的三种方式前面示例中使用的是命令式imperative风格直接调用app.router.add_get(...)等。另外还有两种等价方式路由表route tables与路由装饰器route decorators。路由表类似 Django 风格async def handle_get(request): ... async def handle_post(request): ... app.router.add_routes([web.get(/get, handle_get), web.post(/post, handle_post)])该片段调用UrlDispatcher.add_routes注册一组route definitions即aiohttp.web.RouteDef实例它们由aiohttp.web.get或aiohttp.web.post等函数创建。RouteDef.register会把标准方法在hdrs.METH_ALL内分派到router.add_method()其他方法走router.add_route()见 aiohttp/web_routedef.py。路由装饰器更接近 Flask 风格routes web.RouteTableDef() routes.get(/get) async def handle_get(request): ... routes.post(/post) async def handle_post(request): ... app.router.add_routes(routes)装饰器同样可以用于基于类的视图routes web.RouteTableDef() routes.view(/view) class MyView(web.View): async def get(self): ... async def post(self): ... app.router.add_routes(routes)这个例子首先创建了一个aiohttp.web.RouteTableDef容器——它是一个类列表对象附带RouteTableDef.get、RouteTableDef.post等用于注册新路由的装饰器。容器填充完成后调用UrlDispatcher.add_routes把注册的route definitions加入应用的路由器该 API 自 aiohttp 2.3 起提供。三种方式命令式调用、路由表、装饰器完全等价你可以选择自己喜欢的方式甚至可以混用。JSON 响应返回 JSON 数据是 Web 服务最常见的需求之一aiohttp.web为此提供了快捷函数aiohttp.web.json_responseasync def handler(request): data {some: data} return web.json_response(data)该快捷方法返回的是aiohttp.web.Response实例因此在返回前你还可以给响应设置 cookie 等。从源码看aiohttp/web_response.pyjson_response支持data/text/body三选一同时指定会抛ValueErrorstatus默认 200content_type默认application/json并允许通过dumps参数替换默认的json.dumps序列化器框架还额外提供了面向 orjson 这类返回 bytes 的编码器的json_bytes_response见 aiohttp/web_response.py可避免 str 与 bytes 之间的编解码开销。用户会话Sessions跨请求保存用户数据的需求通常被称为session。aiohttp.web本身没有内置 session 概念但第三方库aiohttp_session提供了完整的 session 支持import asyncio import time import base64 from cryptography import fernet from aiohttp import web from aiohttp_session import setup, get_session, session_middleware from aiohttp_session.cookie_storage import EncryptedCookieStorage async def handler(request): session await get_session(request) last_visit session.get(last_visit) session[last_visit] time.time() text Last visited: {}.format(last_visit) return web.Response(texttext) async def make_app(): app web.Application() # secret_key must be 32 url-safe base64-encoded bytes fernet_key fernet.Fernet.generate_key() secret_key base64.urlsafe_b64decode(fernet_key) setup(app, EncryptedCookieStorage(secret_key)) app.add_routes([web.get(/, handler)]) return app web.run_app(make_app())要点说明EncryptedCookieStorage需要 32 字节的 url-safe base64 密钥示例中通过Fernet.generate_key()生成后再urlsafe_b64decode得到session 数据以加密 cookie 的形式保存在客户端get_session(request)在 handler 中取出一个类 dict 对象可直接读写。注意示例使用了make_app()协程直接传给web.run_app——从源码看_run_app会先await协程类型的app参数见 aiohttp/web.py所以这种写法是受支持的。HTTP 表单处理HTTP Forms 开箱即用。如果表单方法是GETform methodget使用aiohttp.web.BaseRequest.query获取表单数据——它返回 URL 查询字符串解析出的MultiDictProxy见 aiohttp/web_request.py。如果表单方法为POST使用aiohttp.web.BaseRequest.post或aiohttp.web.BaseRequest.multipart。BaseRequest.post同时接受application/x-www-form-urlencoded与multipart/form-data两种表单编码例如form enctypemultipart/form-data。它会把文件数据暂存到临时目录。如果请求体超过client_max_size或者表单字段数超过client_max_fields默认 1000设为0可禁用上限post会抛出aiohttp.web.HTTPRequestEntityTooLarge异常。这两个限制的默认值分别为1024 ** 21 MiB与1000可在Application构造时通过同名参数调整见 aiohttp/web_app.py。出于效率考虑推荐使用BaseRequest.multipart它在上传大文件时尤其高效见下文文件上传。例如下面的表单form action/login methodpost accept-charsetutf-8 enctypeapplication/x-www-form-urlencoded label forloginLogin/label input idlogin namelogin typetext value autofocus/ label forpasswordPassword/label input idpassword namepassword typepassword value/ input typesubmit valuelogin/ /form提交的值可以这样读取async def do_login(request): data await request.post() login data[login] password data[password]文件上传aiohttp.web内置了对浏览器上传文件的处理支持。首先确保 HTMLform元素的enctype属性设置为enctypemultipart/form-data。例如下面是一个接收 MP3 文件的表单form action/store/mp3 methodpost accept-charsetutf-8 enctypemultipart/form-data label formp3Mp3/label input idmp3 namemp3 typefile value/ input typesubmit valuesubmit/ /form然后在请求处理器中可以把文件输入字段作为FileField实例访问。FileField是文件本身及其部分元数据的容器async def store_mp3_handler(request): # WARNING: dont do that if you plan to receive large files! data await request.post() mp3 data[mp3] # .filename contains the name of the file in string format. filename mp3.filename # .file contains the actual file data that needs to be stored somewhere. mp3_file data[mp3].file content mp3_file.read() return web.Response(bodycontent, headersMultiDict( {CONTENT-DISPOSITION: mp3_file}))注意上面示例中的大警告问题在于BaseRequest.post会把整个 payload 读入内存可能导致 :abbr:OOMOut Of Memory错误。为规避这一点multipart 上传应改用BaseRequest.multipart它返回一个 multipart readerasync def store_mp3_handler(request): reader await request.multipart() # /!\ Dont forget to validate your inputs /!\ # reader.next() will yield the fields of your form field await reader.next() assert field.name name name await field.read(decodeTrue) field await reader.next() assert field.name mp3 filename field.filename # You cannot rely on Content-Length if transfer is chunked. size 0 with open(os.path.join(/spool/yarrr-media/mp3/, filename), wb) as f: while True: chunk await field.read_chunk() # 8192 bytes by default. if not chunk: break size len(chunk) f.write(chunk) return web.Response(text{} sized of {} successfully stored .format(filename, size))要点reader.next()依次产出表单的各个字段BodyPartReader.read_chunk()按块读取内容默认块大小为 8192 字节DEFAULT_CHUNK_SIZE逐块写入磁盘文件内存占用保持恒定因此即使文件远大于client_max_size也能安全流式落地multipart 相关实现细节可参考 aiohttp/multipart.py 与 docs/multipart.rst。值得一提的是request.post()内部对 multipart 文件实际使用的是SpooledTemporaryFile小于 1 MiB_FILE_SPOOL_MAX_SIZE的数据驻留内存超限后自动滚落到磁盘临时文件见 aiohttp/web_request.py这也是它能承受比client_max_size更大的文件的原因——不过它仍然会在整体尺寸上受client_max_size约束。WebSocketsaiohttp.web对WebSockets提供了开箱即用的支持。要建立一个 WebSocket在请求处理器中创建WebSocketResponse并用它和客户端通信async def websocket_handler(request): ws web.WebSocketResponse() await ws.prepare(request) async for msg in ws: # ws.__next__() automatically terminates the loop # after ws.close() or ws.exception() is called if msg.type aiohttp.WSMsgType.TEXT: if msg.data close: await ws.close() else: await ws.send_str(msg.data /answer) elif msg.type aiohttp.WSMsgType.ERROR: print(ws connection closed with exception %s % ws.exception()) print(websocket connection closed) return ws该 handler 应注册为 HTTP GET 处理器app.add_routes([web.get(/ws, websocket_handler)])机制说明await ws.prepare(request)完成 HTTP 升级握手此后async for msg in ws会持续产出消息msg.type用WSMsgType枚举区分文本TEXT、错误ERROR等类型当ws.close()被调用或连接出现异常时循环自动终止ws.exception()可取回异常对象。源码中WebSocketResponse继承自StreamResponse还提供ping()/pong()心跳、send_str()等方法见 aiohttp/web_ws.py。完整的可运行示例参见 examples/web_ws.py 与 examples/websocket.html握手细节可查阅 docs/web_reference.rst 与 docs/websocket_utilities.rst。重定向要把用户重定向到另一个端点抛出HTTPFound即可其location可以是绝对 URL、相对 URL 或视图名router 中的参数raise web.HTTPFound(/redirect)下面的例子演示了重定向到路由表中名为login的视图async def handler(request): location request.app.router[login].url_for() raise web.HTTPFound(locationlocation) router.add_get(/handler, handler) router.add_get(/login, login_handler, namelogin)一个带登录校验的完整示例aiohttp_jinja2.template(login.html) async def login(request): if request.method POST: form await request.post() error validate_login(form) if error: return {error: error} else: # login form is valid location request.app.router[index].url_for() raise web.HTTPFound(locationlocation) return {} app.router.add_get(/, index, nameindex) app.router.add_get(/login, login, namelogin) app.router.add_post(/login, login, namelogin)可以看到重定向配合命名路由的url_for()见上文使用命名资源反向构造 URL可以完全避免在业务代码中硬编码路径——即使路由路径日后调整所有重定向与模板 URL 都会自动跟随。HTTP 异常体系HTTPFound是 302还有HTTPMovedPermanently301、HTTPSeeOther303 等的完整清单见 docs/web_exceptions.rst 与 aiohttp/web_exceptions.py。下一步想深入理解run_app背后的优雅关闭graceful shutdown与应用运行器app runners阅读 docs/web_lowlevel.rst处理复杂应用异步启动、多 host、子应用参考 docs/web_advanced.rst完整的Application、Request、Response、ViewAPI 参考见 docs/web_reference.rst路由定义细节见其中RouteDef/RouteTableDef相关章节更多可直接运行的服务端示例位于 examples/ 目录web_srv.py、web_srv_route_deco.py、web_classview.py、web_ws.py等。【免费下载链接】aiohttpAsynchronous HTTP client/server framework for asyncio and Python项目地址: https://gitcode.com/gh_mirrors/ai/aiohttp创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考