Aiohttp:Server
lastmod 2023-07-27created 2023-07-272 edits
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)
Now add the setup_middlewares step to the main file:
# 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
The best approach is to use a Reverse Proxy (such as nginx or a CDN). If you serve files directly, e.g. for development convenience, use the following.
Or:
Redirects
To redirect user to another endpoint - raise HTTPFound with an absolute URL, relative URL or view name (the argument from router):
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
It can be obtained as follows.
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}")
Obtaining values from the middle of the path
# ...
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
DynamicResource
With Variable Resources, when routing with the name name as follows,
you can obtain the path corresponding to name as below.
The logic that assigns to request.match_info lives in the DynamicResource class. To use this feature on its own, do the following:
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
...