aiohttp文档翻译-server(一)
web server 快速入门
运行一个简单的web server
为了实现web server, 首先需要实现request handler
一个 request handler 必须是一个coroutine (协程), 它接受一个request实例作为其唯一参数,并返回一个response 实例,如下代码中的hello
from aiohttp import web async def hello(request): return web.response(text="hello, world")
接下来,创建一个application实例并在特定的http方法和路径上注册请求处理程序
app = web.application() app.add_routes([web.get('/', hello)])
然后,通过run_app() 调用运行应用程序
web.run_app(app)
这样,我们就可以通过http://127.0.0.1:8080 查看结果
或者如果喜欢使用路径装饰器的方式,则创建路由表并注册web处理程序,代码实现方式为:
routes = web.routetabledef() @routes.get('/') async def hello(request): return web.response(text="hello, world") app = web.application() app.add_routes(routes) web.run_app(app)
这两种风格做的是同样的工作,完全看你个人喜好,看你是喜欢django的urls.py风格的还是flask的装饰器风格
handler
请求处理程序必须是一个协程,它接受一个request实例作为其唯一参数,并返回一个streamresponse派生(例如,响应)实例
async def handler(request): return web.response()
handlers 通过使用get() 和post() 等方法将程序在特定路由(http方法和路径对) 上使用application.add_routes()注册它们来处理请求
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)
route()或routetabledef.route()也支持通配符http方法,允许处理程序在具有任何http方法的路径上提供传入请求
app.add_routes[web.route('*', '/path', all_handler)]
默认情况下,使用get方法添加的端点将接受head请求并返回与get请求相同的响应头。您还可以拒绝路由上的head请求:
web.get('/', handler, allow_head=false)
这里不会在head请求上调用处理程序,服务器将响应405:method not allowed。
resources and routes
内部路由由application.router(urldispatcher实例)提供。
路由器是资源列表
资源是路由表中的条目,对应于请求的url。
资源至少有一条路由
route对应于通过调用web处理程序来处理http方法。
这个库实现合并相同路径的路由,为所有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)]
variable resources
资源也可能具有可变路径。例如,路径为“/ a / {name} / c”的资源会匹配所有传入请求,其路径为“/ a / b / c”,“/ a / 1 / c”和“/ a /”
变量部分以{identifier}形式指定,其中标识符稍后可以在请求处理程序中使用,以访问该部分的匹配值。这是通过在request.match_info映射中查找标识符来完成的:
@routes.get('/{name}') async def variable_handler(request): return web.response( text="hello, {}".format(request.match_info['name']))
默认情况下,每个部分都匹配正则表达式[^ {} /] +。你还可以使用{identifier:regex}格式指定自定义正则表达式:
web.get(r'/{name:\d+}', handler)
reverse url constructing using named resources
routes也可以给出一个名字:
@routes.get('/root', name='root') async def handler(request): ...
然后可以使用它来访问和构建该资源的url(例如在请求处理程序中):
url == request.app.router['root'].url_for().with_query({"a": "b", "c": "d"}) assert url == url('/root?a=b&c=d')
一个更有趣的例子是为可变资源构建url:
app.router.add_resource(r'/{user}/info', name='user-info')
在这种情况下,你还可以传递路线的各个部分:
url = request.app.router['user-info'].url_for(user='john_doe') url_with_qs = url.with_query("a=b") assert url_with_qs == '/john_doe/info?a=b'
organizing handlers in classes
class handler: def __init__(self): pass async def handle_intro(self, request): return web.response(text="hello, world") async def handle_greeting(self, request): name = request.match_info.get('name', "anonymous") txt = "hello, {}".format(name) return web.response(text=txt) handler = handler() app.add_routes([web.get('/intro', handler.handle_intro), web.get('/greet/{name}', handler.handle_greeting)]
class based views
aiohttp.web支持基于类的视图。你可以从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)
处理程序应该是coroutines只接受self并返回响应对象作为常规web处理程序。view.request属性可以检索请求对象。实现视图后(上面的例子中的myview)应该在应用程序的路由器中注册:
web.view('/path/to', myview)
或者
@routes.view('/path/to') class myview(web.view): ...
resource views
可以使用urldispatcher.resources()方法查看路由器中的所有注册资源
for resource in app.router.resources(): print(resource)
可以使用urldispatcher.named_resources()方法查看使用名称注册的资源的子集:
for name, resource in app.router.named_resources().items(): print(name, resource)
alternative ways for registering routes
上面显示的代码示例使用命令式样式添加新路由:它们调用app.router.add_get(...)等。有两种选择:路由表和路由装饰器。路由表看起来像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),
该片段调用add_routes()来注册由aiohttp.web.get()或aiohttp.web.post()函数创建的路由定义列表(aiohttp.web.routedef实例)。
路由装饰器和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容器。容器是一个类似列表的对象,带有额外的装饰器aiohttp.web.routetabledef.get(),aiohttp.web.routetabledef.post()等,用于注册新路由。填充容器后,add_routes()用于将已注册的路由定义添加到应用程序的路由器中。
json response
返回json数据是一种常见的情况,aiohttp.web提供了返回json的快捷方式 - aiohttp.web.json_response():
async def handler(request): data = {'some': 'data'} return web.json_response(data)
这个快捷方法返回aiohttp.web.response实例,因此你可以在从处理程序返回cookie之前设置cookie。
user sessions
通常,您需要一个容器来跨请求存储用户数据。该概念通常称为会话。aiohttp.web没有内置的会话概念,但是,有一个第三方库aiohttp_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['last_visit'] if 'last_visit' in session else none text = 'last visited: {}'.format(last_visit) return web.response(text=text) 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())
http forms
开箱即用支持http表单。
如果form的方法是“get”(<form method =“get”>),请使用request.query获取表单数据。要使用“post”方法访问表单数据,请使用request.post()或request.multipart()。
request.post()接受'application / x-www-form-urlencoded'和'multipart / form-data'表单的数据编码(例如<form enctype =“multipart / form-data”>)。它将文件数据存储在临时目录中。如果在引发valueerror异常后指定了client_max_size。
为了提高效率,请使用request.multipart(),它对于上传大文件(文件上传)特别有效。
通过以下表格提交的值:
<form action="/login" method="post" accept-charset="utf-8" enctype="application/x-www-form-urlencoded"> <label for="login">login</label> <input id="login" name="login" type="text" value="" autofocus/> <label for="password">password</label> <input id="password" name="password" type="password" value=""/> <input type="submit" value="login"/> </form>
async def do_login(request): data = await request.post() login = data['login'] password = data['password']
file uploads
aiohttp.web内置支持处理从浏览器上传的文件。首先,确保html <form>元素的enctype属性设置为enctype =“multipart / form-data”。
例如,这是一个接受mp3文件的表单:
<form action="/store/mp3" method="post" accept-charset="utf-8" enctype="multipart/form-data"> <label for="mp3">mp3</label> <input id="mp3" name="mp3" type="file" value=""/> <input type="submit" value="submit"/> </form>
然后,在请求处理程序中,你可以将文件输入字段作为filefield实例进行访问。filefield只是文件的容器及其一些元数据:
async def store_mp3_handler(request): # warning: don't 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(body=content, headers=multidict( {'content-disposition': mp3_file}))
您可能已经注意到上面示例中的一个重要警告。一般问题是request.post()读取内存中的整个有效负载,导致可能的oom错误。为避免这种情况,对于分段上传,您应该使用request.multipart()来返回多部分阅读器:
async def store_mp3_handler(request): reader = await request.multipart() # /!\ don't 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(decode=true) 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))
websockets
aiohttp.web支持websockets开箱即用。要设置websocket,请在请求处理程序中创建websocketresponse,然后使用它与对等方进行通信:
async def websocket_handler(request): ws = web.websocketresponse() await ws.prepare(request) async for msg in ws: 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
处理程序应注册为http get处理器
app.add_routes([web.get('/ws', websocket_handler)])
redirects
将用户重定向到另一个端点 - 使用绝对url,相对url或视图名称(来自路由器的参数)引发httpfound:
raise web.httpfound('/redirect')
以下示例显示重定向到路径中名为'login'的视图:
async def handler(request): location = request.app.router['login'].url_for() raise web.httpfound(location=location) router.add_get('/handler', handler) router.add_get('/login', login_handler, name='login')
登录验证示例
@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(location=location) return {} app.router.add_get('/', index, name='index') app.router.add_get('/login', login, name='login') app.router.add_post('/login', login, name='login')
推荐阅读
-
Python核心模块urllib的学习(一)--翻译官方Python文档urllib.request
-
Spark官方文档翻译(一)~Overview
-
aiohttp文档翻译-server(一)
-
OAuth2 Boot(一、授权服务器)(官方文档翻译)
-
再见所有的翻译工具!我使用了有道智云+Python开发一个批量文档翻译工具,老大对我赞不绝口
-
Python核心模块urllib的学习(一)--翻译官方Python文档urllib.request
-
【翻译】SQL Server索引进阶:第一级,索引简介
-
Spark官方文档翻译(一)~Overview
-
《JBoss Seam:一个深度集成框架》中文翻译文档
-
aiohttp文档翻译-server(一)