TopolearnDatabase pools and HTTP clients must be created once at startup and closed at shutdown, not per request.
An Application is a mapping you can store shared objects in, such as database pools, a ClientSession for outgoing calls, or caches. Since aiohttp 3.9 the recommended keys are typed web.AppKey objects:
db_key = web.AppKey("db", asyncpg.Pool)
app[db_key] = pool
pool = request.app[db_key] # type checkers know this is a PoolString keys still work but emit a warning and lose type information.
Resources are created and destroyed via lifecycle signals:
app.cleanup_ctx: async generators that run setup code beforeyieldat startup and teardown afteryieldat cleanup. This is the preferred style because setup and teardown live together and teardown only runs if setup succeeded.app.on_startup,app.on_shutdown,app.on_cleanup: lists of coroutines called at each phase.on_shutdownruns first when stopping, a good time to close WebSockets; thenon_cleanup.
After the app starts it is frozen: adding routes, middlewares or signals then raises an error. Modifying state values (app[key]) stays allowed.
Large apps can be split into sub-applications with app.add_subapp("/admin/", admin_app), each with its own middlewares and state, and nested apps reach the parent's state through request.config_dict, which looks keys up through the chain.
Per-request data goes on the request (request["user"]), never on the app, because the app is shared by all concurrent requests.
Example
import asyncio
import aiohttp
from aiohttp import web
from aiohttp.test_utils import TestClient, TestServer
http_key = web.AppKey("http", aiohttp.ClientSession)
events = []
async def http_client(app):
app[http_key] = aiohttp.ClientSession() # setup at startup
events.append("started")
yield
await app[http_key].close() # teardown at cleanup
events.append("closed")
async def status(request):
return web.json_response({"open": not request.app[http_key].closed})
async def demo():
app = web.Application()
app.cleanup_ctx.append(http_client)
app.router.add_get("/status", status)
async with TestClient(TestServer(app)) as c:
print(await (await c.get("/status")).json()) # {'open': True}
print(events) # ['started', 'closed']
asyncio.run(demo())Remember
- Store shared resources in the app with typed web.AppKey keys
- cleanup_ctx async generators pair setup and teardown
- The app freezes at startup; per-request data goes on the request
Watch out for
- Creating a ClientSession or DB pool inside each handler instead of once at startup.