TopolearnReal endpoints take input; reading and validating it correctly prevents both bugs and security holes.
web.Request gives access to everything the client sent:
request.match_info: path variables from the route.request.query: query parameters as a read-only multidict of strings;request.query.get("page", "1"),getall("tag")for repeated keys.request.headers,request.cookies,request.method,request.path,request.url,request.remote.- The body, read with coroutines:
await request.json(),await request.post()(form fields and uploaded files),await request.text(),await request.read(), orrequest.contentas a stream for large uploads. request.can_read_body/request.body_existsto check whether there is a body.
All input is untrusted. Convert and validate explicitly: int(request.query["page"]) can raise ValueError, and await request.json() raises json.JSONDecodeError on bad JSON. Translate such errors into web.HTTPBadRequest rather than letting them become 500s. Many projects use Pydantic models for validation.
aiohttp enforces a maximum body size, client_max_size on the Application (default 1 MiB), and answers 413 Payload Too Large above it. Raise it for upload endpoints, or stream request.content.
For responses, web.json_response(data, status=201) sets the status, and you can add headers via headers=. raise web.HTTPBadRequest(text="...") (or reason=) ends the request with a 400. For JSON error bodies, pass text=json.dumps(...) with content_type="application/json".
Example
import asyncio
import json
from aiohttp import web
from aiohttp.test_utils import TestClient, TestServer
async def create_item(request):
try:
payload = await request.json()
qty = int(payload["qty"])
except (json.JSONDecodeError, KeyError, ValueError, TypeError):
raise web.HTTPBadRequest(text="expected JSON with integer qty")
dry_run = request.query.get("dry_run") == "1"
return web.json_response({"qty": qty, "dry_run": dry_run}, status=201)
app = web.Application(client_max_size=2 * 1024**2) # allow 2 MiB bodies
app.router.add_post("/items", create_item)
async def demo():
async with TestClient(TestServer(app)) as c:
r = await c.post("/items?dry_run=1", json={"qty": "3"})
print(r.status, await r.json()) # 201 {'qty': 3, 'dry_run': True}
r = await c.post("/items", data="not json")
print(r.status, await r.text()) # 400 expected JSON with integer qty
asyncio.run(demo())Remember
- match_info for path, query for ?params, await json()/post()/read() for bodies
- Validate input and raise web.HTTPBadRequest instead of failing with 500
- client_max_size limits body size (default 1 MiB)
Watch out for
- Letting int(request.query['x']) raise ValueError, which becomes a 500 error.