Skip to content

Aiohttp:Server

Middlewares 

from aiohttp.web import middleware

@middleware
async def middleware(request, handler):
    resp = await handler(request)
    resp.text = resp.text + ' wink'
    return resp

Handling error pages 

# middlewares.py
import aiohttp_jinja2
from aiohttp import web


async def handle_404(request):
    return aiohttp_jinja2.render_template('404.html', request, {}, status=404)


async def handle_500(request):
    return aiohttp_jinja2.render_template('500.html', request, {}, status=500)


def create_error_middleware(overrides):

    @web.middleware
    async def error_middleware(request, handler):
        try:
            return await handler(request)
        except web.HTTPException as ex:
            override = overrides.get(ex.status)
            if override:
                return await override(request)

            raise
        except Exception:
            return await overrides[500](request)

    return error_middleware


def setup_middlewares(app):
    error_middleware = create_error_middleware({
        404: handle_404,
        500: handle_500
    })
    app.middlewares.append(error_middleware)

이제 setup_middlewares기본 파일에 단계를 추가 하십시오: 

# aiohttpdemo_polls/main.py
from aiohttp import web

from settings import config
from routes import setup_routes
from middlewares import setup_middlewares

app = web.Application()
setup_routes(app)
setup_middlewares(app)
app['config'] = config
web.run_app(app)

Variable Resources 

@routes.get('/{name}')
async def variable_handler(request):
    return web.Response(
        text="Hello, {}".format(request.match_info['name']))

Static file handling

가장 좋은 방법은 Reverse Proxy를 사용(nginx 또는 CDN 같은)하는 것이다. 개발의 편의성 등을 위해 직접 사용할 경우 다음과 같이 사용한다. 

app.add_routes([web.static('/prefix', path_to_static_folder)])

또는: 

routes.static('/prefix', path_to_static_folder)

Redirects

To redirect user to another endpoint - raise HTTPFound with an absolute URL, relative URL or view name (the argument from router): 

raise web.HTTPFound('/redirect')

The following example shows redirect to view named ‘login’ in routes: 

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')

Obtain Remote IP address 

from aiohttp import web

async def on_credentials(request):
    name = request.match_info.get('name', "Anonymous")
    text = "Hello, " + name
    peername = request.transport.get_extra_info('peername')
    if peername is not None:
        host, port = peername
        print(f"host={host},port={port}")
    return web.Response(text=text)

app = web.Application()
app.add_routes([web.get('/credentials', on_credentials)])

if __name__ == '__main__':
    web.run_app(app)

GET Method URL query

다음과 같이 획득할 수 있다. 

async def get_handler(self, request):

    param1 = request.rel_url.query.get('name', '')
    param2 = request.rel_url.query.get('age', '')

    return web.Response(text=f"Name: {param1}, Age: {param2}")

중간, 경로 값 획득 

    # ...
    web.get("/project/{proj}/graph", self.on_get_graph)
    #...

    async def on_get_graph(self, request: Request):
        projname = request.match_info["proj"]
        return web.Response()

Request body to json 

async def get_handler(self, request):
    json = await request.json()
    # ...

DynamicResource

Variable Resources을 사용하면, 다음과 같이 name이름을 라우팅 할 때, 

web.get(r'/{name:\d+}', handler)

아래와 같이 name 에 해당하는 경로를 획득할 수 있다. 

async def route(self, request: Request):
    name = request.match_info["name"]

이 때, request.match_info 에 할당하는 로직은 DynamicResource 클래스에 있다. 이 기능만 따로 사용하고 싶다면 다음과 같이 사용하면 된다: 

from aiohttp.web_urldispatcher import DynamicResource
res = DynamicResource("/{temp}/{tail:.*}")
result = res._match("/kkk/unknown/mmm")  # noqa

Socket initialize 

sock = socket.socket()
with contextlib.closing(sock):
    sock.bind(("0.0.0.0", 0))
    _, port = sock.getsockname()
    web.run_app(app, sock=sock, print=printer)
    patched_loop.create_server.assert_called_with(
        mock.ANY, sock=sock, backlog=128, ssl=None
    )
    assert f"http://0.0.0.0:{port}" in printer.call_args[0][0]

Long polling

Graceful exit 

import asyncio
from aiohttp import web

app = web.Application()
routes = web.RouteTableDef()

@routes.get("/hello")
async def hello(request):
    return web.Response(text="Hello World")


async def bootup(this_app):
    asyncio.create_task(background())

async def background():
    await asyncio.sleep(5)
    print("Start shutting down")
    await app.shutdown()
    print("Start cleaning up")
    await app.cleanup()

if __name__ == "__main__":
    app.add_routes(routes)
    app.on_startup.append(bootup)
    web.run_app(app, host="0.0.0.0", port=80)
    print("Finished")

Server Deployment

Nginx+supervisord

...

See also