aiohttp-request 0.0.2

安装

pip install aiohttp_request

使用

import asyncio

from aiohttp import web
from aiohttp_request import ThreadContext, middleware_factory, grequest, get_request


def thread():
    assert grequest['sense'] == 42


async def task():
    # grequest is `lazy` version of request
    assert grequest['sense'] == 42

    loop = asyncio.get_event_loop()
    # works for threads as well with ThreadContext
    await loop.run_in_executor(None, ThreadContext(thread))


async def hello(request):
    # get_request is on-demand function to get current request
    assert get_request() is request

    request['sense'] = 42

    # asyncio.Task is supported
    await asyncio.ensure_future(task())

    return web.Response(text="Hello, world")


app = web.Application(middlewares=[middleware_factory()])
app.add_routes([web.get('/', hello)])
web.run_app(app)

需要Python 3.7+，无法支持旧版本的Python!!!

注意

该库依赖于PEP 567及其asyncio支持

aiohttp-request通过contextvars_executor与线程配合良好，无需ThreadContext

import asyncio

from aiohttp import web
from aiohttp_request import middleware_factory, grequest
from contextvars_executor import ContextVarExecutor


def thread():
    assert grequest['sense'] == 42


async def hello(request):
    request['sense'] = 42

    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, thread)

    return web.Response(text="Hello, world")


loop = asyncio.get_event_loop()
loop.set_default_executor(ContextVarExecutor())
app = web.Application(middlewares=[middleware_factory()])
app.add_routes([web.get('/', hello)])
web.run_app(app)