Lesson 2 — read a request, and say no to one¶
What you will build¶
Three routes that read different parts of a request — a path parameter, a query string, and a JSON body — and the errors that refuse one. A handler either returns a value or raises; the useful part is that an error knows which HTTP status it means.
What you need¶
Xyberos with the http extra, from lesson 1.
The code¶
"""Lesson 2 — reading a request, and saying no to one.
Run it with:
python -m learn_traditional.traditional02_requests_and_errors
"""
from __future__ import annotations
from starlette.requests import Request
from xyberos import App, HttpError
from xyberos.http import HttpAdapter, read_json_object
from xyberos.http.testing import create_test_client
ITEMS = {1: "notebook"}
class ItemNotFoundError(HttpError):
"""A domain error that knows which HTTP status it means."""
def __init__(self, item_id: int) -> None:
super().__init__(f"item {item_id} was not found", status_code=404)
def create_app() -> App:
"""Build the application."""
app = App()
@app.get("/items/{item_id}")
async def get_item(item_id: int) -> dict[str, object]:
if item_id not in ITEMS:
raise ItemNotFoundError(item_id)
return {"item_id": item_id, "name": ITEMS[item_id]}
@app.get("/search")
async def search(request: Request) -> dict[str, str]:
return {"query": request.query_params.get("q", "all")}
@app.post("/items")
async def create_item(request: Request) -> dict[str, object]:
payload = await read_json_object(request)
return {"created": payload.get("name"), "id": max(ITEMS) + 1}
return app
def main() -> None:
with create_test_client(HttpAdapter(create_app())) as client:
print("found:", client.get("/items/1").json())
print("query:", client.get("/search?q=book").json())
missing = client.get("/items/99")
print("missing status:", missing.status_code)
print("missing body:", missing.json())
print("created:", client.post("/items", json={"name": "pen"}).json())
broken = client.post("/items", content=b"this is not json")
print("broken body status:", broken.status_code)
print("broken body type:", broken.json()["error"]["type"])
if __name__ == "__main__":
main()
How to run it¶
What it prints¶
found: {'item_id': 1, 'name': 'notebook'}
query: {'query': 'book'}
missing status: 404
missing body: {'error': {'type': 'ItemNotFoundError', 'message': 'item 99 was not found'}}
created: {'created': 'pen', 'id': 2}
broken body status: 422
broken body type: RequestValidationError
Read the pairs. missing status and missing body are the same response: a 404 whose body names the
error type and its message. broken body status and broken body type are another: a 422 from
read_json_object, reported as RequestValidationError because the client sent something wrong.
Reading input, and refusing it¶
| You raise | The client receives |
|---|---|
a subclass of HttpError |
its status_code and {"error": {"type", "message"}} |
RequestValidationError |
422, named as a validation failure |
| anything else | 500 with a generic message — the traceback goes to the log, never to the client |
There is one recommended way to read a body: payload = await read_json_object(request). It
returns a dictionary or raises RequestValidationError, so a handler never parses bytes and decides
for itself what a malformed body means. Everything else you read from request directly:
request.query_params, request.headers, request.path_params.
The rule that matters¶
A request failed because of the client only when you say so. A 500 is your bug, and the client
learns nothing about it beyond the status code. That is why the exception above is a 404 with a
message written for a person, and why an unexpected error never leaks its traceback.
Common mistakes¶
- Raising
HTTPExceptionfrom Starlette. It works, because the adapter maps it, but it puts a web framework's type in your service layer. A domain error that derives fromHttpErrorkeeps the status code where the domain decision is. - Returning an error instead of raising it. A handler that returns
{"error": ...}sends200, and every client-side retry policy is now wrong. - Treating
X-Request-IDas identity. It is echoed for log correlation and is supplied by the client, so anything security-related must not read it — lesson 8 is where a real principal arrives.
Exercise¶
Return 409 Conflict when an item is created twice.
Show the answer
"""Lesson 2, exercise — say no with the right status code.
The answer to the lesson's exercise: creating the same item twice is a conflict, and the status code
says so without the handler thinking about HTTP beyond the number.
"""
from __future__ import annotations
from starlette.requests import Request
from xyberos import App, HttpError
from xyberos.http import HttpAdapter, read_json_object
from xyberos.http.testing import create_test_client
class ItemAlreadyExistsError(HttpError):
"""A domain error that knows which HTTP status it means."""
def __init__(self, name: str) -> None:
super().__init__(f"{name!r} already exists", status_code=409)
def create_app() -> App:
"""Build the application, with the store created per application rather than per module."""
items = {1: "notebook"}
app = App()
@app.post("/items")
async def create_item(request: Request) -> dict[str, object]:
payload = await read_json_object(request)
name = str(payload.get("name"))
if name in items.values():
raise ItemAlreadyExistsError(name)
item_id = max(items) + 1
items[item_id] = name
return {"created": name, "id": item_id}
return app
def main() -> None:
"""Create one item, then try to create it again."""
with create_test_client(HttpAdapter(create_app())) as client:
print("created:", client.post("/items", json={"name": "pen"}).json())
conflict = client.post("/items", json={"name": "pen"})
print("conflict status:", conflict.status_code)
print("conflict type:", conflict.json()["error"]["type"])
if __name__ == "__main__":
main()
Next¶
Lesson 3 stores the items in SQLite instead of a dictionary, and introduces the transaction that makes "write, then fail" safe. Until it is published, the roadmap shows where each lesson stands.