Topolearnaiohttp's server side is a compact async web framework suitable for APIs, proxies and WebSocket services.
The server lives in aiohttp.web. A handler is a coroutine that takes a web.Request and returns a web.Response:
async def hello(request):
return web.Response(text="Hello")Register handlers on a web.Application, either with app.add_routes([web.get("/", hello), web.post("/items", create)]) or with a web.RouteTableDef, which provides Flask-like decorators: @routes.get("/"). Then start the server with web.run_app(app, host="127.0.0.1", port=8080), which runs the event loop until interrupted.
Responses:
web.Response(text=..., status=..., headers=...)for text, orbody=for bytes,web.json_response(obj, status=...)for JSON,raise web.HTTPNotFound()and the otherweb.HTTPExceptionsubclasses (HTTPBadRequest,HTTPFound(location=...)for redirects) to end a request with an error or redirect.
Path variables are written in braces, /users/{id}, and read with request.match_info["id"] (always a string). A regex can constrain them: /users/{id:\d+}.
Because handlers are coroutines, everything inside must be non-blocking: use async database drivers and await I/O. Calling blocking code (a synchronous DB driver, time.sleep, heavy CPU work) stalls every other request; push such work to loop.run_in_executor or asyncio.to_thread.
Example
import asyncio
from aiohttp import web
from aiohttp.test_utils import TestClient, TestServer
routes = web.RouteTableDef()
@routes.get("/")
async def index(request):
return web.Response(text="Hello, aiohttp")
@routes.get(r"/users/{id:\d+}")
async def get_user(request):
if request.match_info["id"] != "1":
raise web.HTTPNotFound(text="no such user")
return web.json_response({"id": 1, "name": "Ada"})
app = web.Application()
app.add_routes(routes)
# web.run_app(app, port=8080) # serve for real at http://localhost:8080/
async def demo(): # or exercise it in-process:
async with TestClient(TestServer(app)) as client:
print(await (await client.get("/")).text()) # Hello, aiohttp
print(await (await client.get("/users/1")).json()) # {'id': 1, 'name': 'Ada'}
print((await client.get("/users/2")).status) # 404
asyncio.run(demo())Remember
- Handlers are coroutines: Request in, Response out
- RouteTableDef decorators or app.add_routes register routes; run_app serves
- Raise web.HTTPNotFound and friends for error responses
Watch out for
- Calling blocking libraries inside handlers, which freezes the whole server.