aiohttp文档翻译-server

Posted zhaof

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了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)

 

以上是关于aiohttp文档翻译-server的主要内容,如果未能解决你的问题,请参考以下文章

为 aiohttp 爬虫注入灵魂

aiohttp web方式提供文件下载服务

aiohttp - 我如何查看 websocket 消息缓冲区?

如何将此 JavaScript 代码片段翻译成 Parenscript?

使用 aiohttp.post 我如何传递一些数据进行迭代

aiohttp--web框架session路由cryptography