Overview

def parse_request_line(line: str) -> tuple[str, str, str]:  # => co-01: request = method + path + version
    parts = line.strip().split(" ")  # => splits "GET /items/5 HTTP/1.1" into 3 whitespace-separated fields
    if len(parts) != 3:  # => a well-formed request line always has exactly 3 space-separated tokens
        raise ValueError(f"malformed request line: {line!r}")  # => not a shape a server should ever accept
    method, path, version = parts  # => unpack in RFC 9110's canonical order: method, target, version
    return method, path, version  # => the caller gets the three parts a handler dispatches on
 
 
line = "GET /items/5 HTTP/1.1"  # => a mocked raw request line, never touching a socket
method, path, version = parse_request_line(line)
print(method, path, version)  # => Output: GET /items/5 HTTP/1.1
 
assert method == "GET"  # => the verb that carries "read" semantics (co-02)
assert path == "/items/5"  # => the resource being addressed, with a path param baked in (co-12)
assert version == "HTTP/1.1"
 
try:
    parse_request_line("not a valid line")  # => only 3 tokens are required; this has 4
    raised = False
except ValueError:  # => malformed input is rejected loudly, not silently mis-parsed
    raised = True
print(raised)  # => Output: True
 
assert raised is True
print("kata-01 OK")

GET /items/5 HTTP/1.1
True
kata-01 OK

def classify_status(code: int) -> str:  # => co-03: RFC 9110 groups status codes into five classes
    if 200 <= code < 300:  # => 2xx: the request succeeded
        return "success"
    if 300 <= code < 400:  # => 3xx: further action (redirect) needed
        return "redirect"
    if 400 <= code < 500:  # => 4xx: the CLIENT made a mistake (bad body, missing auth, wrong path)
        return "client-error"
    if 500 <= code < 600:  # => 5xx: the SERVER failed to fulfil a valid request
        return "server-error"
    raise ValueError(f"not a valid HTTP status code: {code}")  # => outside 200-599 isn't a status class at all
 
 
codes = [200, 201, 204, 301, 400, 401, 404, 405, 422, 500]  # => the exact status set this topic teaches
classes = [classify_status(c) for c in codes]  # => one class label per code, same order
print(classes)  # => Output: ['success', 'success', 'success', 'redirect', 'client-error', 'client-error', 'client-error', 'client-error', 'client-error', 'server-error']
 
assert classify_status(201) == "success"  # => 201 Created -- a successful POST (co-02)
assert classify_status(422) == "client-error"  # => Unprocessable Content, native to RFC 9110 (co-10/co-11)
assert classify_status(500) == "server-error"  # => an unhandled exception, never a stack trace (co-11)
print("kata-02 OK")

['success', 'success', 'success', 'redirect', 'client-error', 'client-error', 'client-error', 'client-error', 'client-error', 'server-error']
kata-02 OK

def validate_required(body: dict[str, object], required: list[str]) -> dict[str, object] | None:  # => co-10/co-11: reject bad shapes with a 422 BEFORE handler logic runs
    missing = [field for field in required if field not in body]  # => every required field not present
    if not missing:  # => nothing missing -- the body is shaped correctly, no error to return
        return None
    return {  # => co-11: a consistent JSON error envelope, never a bare stack trace
        "status": 422,  # => co-03: 422 Unprocessable Content, defined natively in RFC 9110 Section 15.5.21
        "error": {
            "code": "validation_error",
            "message": "missing required field(s)",
            "detail": missing,  # => names EXACTLY which fields were missing, for the caller to fix
        },
    }
 
 
good_body: dict[str, object] = {"title": "Ship report", "owner": "alice"}
bad_body: dict[str, object] = {"title": "Ship report"}  # => "owner" is missing
required_fields = ["title", "owner"]
 
good_result = validate_required(good_body, required_fields)  # => nothing missing
bad_result = validate_required(bad_body, required_fields)  # => "owner" missing
print(good_result)  # => Output: None
print(bad_result)  # => Output: {'status': 422, 'error': {'code': 'validation_error', 'message': 'missing required field(s)', 'detail': ['owner']}}
 
assert good_result is None  # => a fully-shaped body produces no error envelope at all
assert bad_result is not None
assert bad_result["status"] == 422
error = bad_result["error"]
assert isinstance(error, dict) and error["detail"] == ["owner"]  # => names the exact missing field
print("kata-03 OK")

None
{'status': 422, 'error': {'code': 'validation_error', 'message': 'missing required field(s)', 'detail': ['owner']}}
kata-03 OK

DEFAULT_LIMIT = 10  # => co-19: list endpoints page results with a DEFAULTED limit
MAX_LIMIT = 100  # => co-19: and a BOUNDED limit -- callers can't demand the whole table at once
 
 
def parse_pagination(params: dict[str, str]) -> tuple[int, int]:  # => returns (limit, offset)
    raw_limit = params.get("limit")  # => absent when the caller sends no ?limit= at all
    raw_offset = params.get("offset")  # => absent when the caller sends no ?offset= at all
 
    limit = int(raw_limit) if raw_limit is not None else DEFAULT_LIMIT  # => fall back to the default
    offset = int(raw_offset) if raw_offset is not None else 0  # => a page always starts at 0 by default
 
    if limit < 0 or offset < 0:  # => negative values make no sense as a window into a list
        raise ValueError("limit and offset must be non-negative")
    limit = min(limit, MAX_LIMIT)  # => co-19: clamp an over-large request instead of scanning everything
 
    return limit, offset
 
 
defaults = parse_pagination({})  # => neither param supplied
explicit = parse_pagination({"limit": "25", "offset": "50"})  # => a normal page request
clamped = parse_pagination({"limit": "9999", "offset": "0"})  # => an abusive over-large limit
 
print(defaults)  # => Output: (10, 0)
print(explicit)  # => Output: (25, 50)
print(clamped)  # => Output: (100, 0) -- clamped down to MAX_LIMIT, never trusted as-is
 
assert defaults == (10, 0)
assert explicit == (25, 50)
assert clamped == (100, 0)
 
try:
    parse_pagination({"limit": "-5"})  # => a negative limit is rejected outright
    raised = False
except ValueError:
    raised = True
print(raised)  # => Output: True
assert raised is True
print("kata-04 OK")

(10, 0)
(25, 50)
(100, 0)
True
kata-04 OK

def build_unsafe_query(status: str) -> str:  # => co-14/co-20: string-concatenated -- the INJECTION-UNSAFE way
    return f"SELECT * FROM tasks WHERE status = '{status}'"  # => user input lands DIRECTLY in the SQL text
 
 
def build_safe_query(status: str) -> tuple[str, tuple[str, ...]]:  # => co-14: parameterized -- the SAFE way
    return "SELECT * FROM tasks WHERE status = ?", (status,)  # => the driver binds params, never concatenates
 
 
def is_injection_attempt(user_input: str) -> bool:  # => a crude but effective mocked detector for this kata
    suspicious = ["'", "--", ";", " OR ", " or "]  # => classic single-quote breakout + comment + stacking
    return any(token in user_input for token in suspicious)
 
 
malicious_input = "done' OR '1'='1"  # => a classic SQL-injection payload targeting the status filter
 
unsafe_sql = build_unsafe_query(malicious_input)  # => the payload becomes part of the SQL TEXT itself
safe_sql, safe_params = build_safe_query(malicious_input)  # => the payload stays DATA, never SQL syntax
 
print(unsafe_sql)  # => Output: SELECT * FROM tasks WHERE status = 'done' OR '1'='1'
print(safe_sql, safe_params)  # => Output: SELECT * FROM tasks WHERE status = ? ("done' OR '1'='1",)
 
# => the unsafe query's WHERE clause now ALWAYS matches every row -- the injection succeeded structurally
assert "OR '1'='1'" in unsafe_sql  # => the payload escaped the intended string literal boundary
# => the safe query's placeholder never changes shape -- the payload is inert, quoted DATA, not code
assert safe_sql == "SELECT * FROM tasks WHERE status = ?"  # => structurally IDENTICAL regardless of input
assert safe_params == (malicious_input,)  # => the whole payload is just one opaque bound parameter
 
assert is_injection_attempt(malicious_input) is True  # => flags the suspicious payload
assert is_injection_attempt("done") is False  # => a normal, benign filter value passes through clean
print("kata-05 OK")

SELECT * FROM tasks WHERE status = 'done' OR '1'='1'
SELECT * FROM tasks WHERE status = ? ("done' OR '1'='1",)
kata-05 OK

def check_session(  # => co-17: server-side session -- the STORE, not the cookie, is the source of truth
    session_id: str | None, sessions: dict[str, str]
) -> str | None:
    if session_id is None:  # => no cookie sent at all
        return None
    return sessions.get(session_id)  # => O(1) lookup: cookie is only an opaque KEY into server-side state
 
 
def check_bearer_token(  # => co-18: a stateless bearer token -- the token itself carries the identity
    authorization: str | None, valid_tokens: dict[str, str]
) -> tuple[int, str | None]:  # => returns (status, caller) -- no session store consulted at all
    if authorization is None or not authorization.startswith("Bearer "):  # => co-18: missing/malformed
        return 401, None  # => reject BEFORE ever touching valid_tokens
    token = authorization.removeprefix("Bearer ")  # => strip the scheme, keep the opaque token value
    caller = valid_tokens.get(token)  # => co-18: reads Authorization, validates it against known tokens
    if caller is None:  # => the token doesn't match any known-valid entry
        return 401, None
    return 200, caller  # => a good token identifies the caller with no server-side session lookup
 
 
sessions: dict[str, str] = {"sess-abc123": "alice"}  # => co-17: session store -- state lives on the SERVER
tokens: dict[str, str] = {"tok-xyz789": "bob"}  # => co-17/co-18: token store -- state lives IN the token
 
session_hit = check_session("sess-abc123", sessions)  # => cookie resolves via the server-side store
session_miss = check_session("sess-nope", sessions)  # => an unknown/expired session id
print(session_hit, session_miss)  # => Output: alice None
 
status_ok, caller_ok = check_bearer_token("Bearer tok-xyz789", tokens)  # => a valid bearer token
status_missing, caller_missing = check_bearer_token(None, tokens)  # => co-18: no Authorization header at all
status_bad, caller_bad = check_bearer_token("Bearer tok-wrong", tokens)  # => co-18: a malformed/unknown token
print(status_ok, caller_ok)  # => Output: 200 bob
print(status_missing, caller_missing)  # => Output: 401 None
print(status_bad, caller_bad)  # => Output: 401 None
 
assert session_hit == "alice" and session_miss is None  # => the cookie alone means nothing without the store
assert (status_ok, caller_ok) == (200, "bob")  # => the token alone is enough -- no store lookup needed
assert status_missing == 401 and status_bad == 401  # => co-18: both reject with 401, not a crash
print("kata-06 OK")

alice None
200 bob
401 None
401 None
kata-06 OK

# => co-02: RFC 9110's idempotent set -- repeating the same call any number of times has the SAME effect
IDEMPOTENT_METHODS: frozenset[str] = frozenset({"GET", "HEAD", "PUT", "DELETE", "OPTIONS", "TRACE"})
 
 
def is_idempotent(method: str) -> bool:  # => O(1) average membership test against the frozen set
    return method.upper() in IDEMPOTENT_METHODS  # => normalize case -- HTTP methods are conventionally upper
 
 
def safe_to_blind_retry(method: str) -> bool:  # => the practical consequence a client relies on
    return is_idempotent(method)  # => co-02: only idempotent methods are safe to retry without side effects
 
 
methods = ["GET", "POST", "PUT", "PATCH", "DELETE"]  # => the five methods this topic teaches
results = {m: is_idempotent(m) for m in methods}  # => one idempotency verdict per method
print(results)  # => Output: {'GET': True, 'POST': False, 'PUT': True, 'PATCH': False, 'DELETE': True}
 
assert results["POST"] is False  # => co-02: POST is explicitly NOT idempotent -- two POSTs can create two rows
assert results["PATCH"] is False  # => RFC 5789: PATCH is "neither safe nor idempotent"
assert results["PUT"] is True  # => PUT "created or replaced" -- repeating it converges to the same state
assert safe_to_blind_retry("GET") is True  # => reads are always safe to retry blindly
assert safe_to_blind_retry("POST") is False  # => a client library must NOT blindly retry a bare POST
print("kata-07 OK")

{'GET': True, 'POST': False, 'PUT': True, 'PATCH': False, 'DELETE': True}
kata-07 OK

from typing import TypedDict
 
 
class ErrorBody(TypedDict):  # => co-11: one FIXED shape every error path returns, never a bare string
    code: str
    message: str
    detail: object | None
 
 
class ErrorEnvelope(TypedDict):  # => the full response body wrapping ErrorBody
    status: int
    error: ErrorBody
 
 
def error_envelope(status: int, code: str, message: str, detail: object | None = None) -> ErrorEnvelope:
    # => co-11: a single constructor every failure path calls -- guarantees UNIFORM shape (co-24: ex-77's idea)
    return {"status": status, "error": {"code": code, "message": message, "detail": detail}}
 
 
not_found = error_envelope(404, "not_found", "task 42 does not exist")  # => co-03: a missing resource
unauthorized = error_envelope(401, "unauthorized", "missing or invalid bearer token")  # => co-18's failure
server_fault = error_envelope(500, "internal_error", "unexpected failure")  # => co-11: NEVER a stack trace
 
print(not_found)  # => Output: {'status': 404, 'error': {'code': 'not_found', 'message': 'task 42 does not exist', 'detail': None}}
print(unauthorized)  # => Output: {'status': 401, 'error': {'code': 'unauthorized', 'message': 'missing or invalid bearer token', 'detail': None}}
print(server_fault)  # => Output: {'status': 500, 'error': {'code': 'internal_error', 'message': 'unexpected failure', 'detail': None}}
 
# => every envelope this function produces shares the EXACT same top-level keys, regardless of status
assert set(not_found.keys()) == set(unauthorized.keys()) == set(server_fault.keys()) == {"status", "error"}
assert not_found["status"] == 404
assert unauthorized["error"]["code"] == "unauthorized"
print("kata-08 OK")

{'status': 404, 'error': {'code': 'not_found', 'message': 'task 42 does not exist', 'detail': None}}
{'status': 401, 'error': {'code': 'unauthorized', 'message': 'missing or invalid bearer token', 'detail': None}}
{'status': 500, 'error': {'code': 'internal_error', 'message': 'unexpected failure', 'detail': None}}
kata-08 OK

from collections.abc import Callable
 
Handler = Callable[[dict[str, str]], str]  # => co-08: a handler takes parsed params, returns a body
 
 
class Router:  # => co-07: maps a method + path PATTERN to a handler function
    def __init__(self) -> None:
        self._routes: list[tuple[str, list[str], Handler]] = []  # => (method, pattern segments, handler)
 
    def add(self, method: str, pattern: str, handler: Handler) -> None:  # => co-07: register one route
        segments = pattern.strip("/").split("/")  # => "/items/{id}" -> ["items", "{id}"]
        self._routes.append((method.upper(), segments, handler))
 
    def dispatch(self, method: str, path: str) -> str:  # => co-07/co-12: match method + path, extract params
        path_segments = path.strip("/").split("/")  # => the ACTUAL incoming path, same segment shape
        for route_method, pattern_segments, handler in self._routes:
            if route_method != method.upper():  # => wrong verb -- this route can't handle this call
                continue
            if len(pattern_segments) != len(path_segments):  # => different segment counts can't match
                continue
            params: dict[str, str] = {}  # => co-12: path params extracted from this specific request
            matched = True
            for pattern_seg, path_seg in zip(pattern_segments, path_segments):  # => compare segment by segment
                if pattern_seg.startswith("{") and pattern_seg.endswith("}"):  # => a PARAM slot, e.g. {id}
                    params[pattern_seg[1:-1]] = path_seg  # => bind the literal path text to its param name
                elif pattern_seg != path_seg:  # => a literal segment that doesn't match -- not this route
                    matched = False
                    break
            if matched:
                return handler(params)  # => co-08: hand the extracted params to the matched handler
        raise LookupError(f"no route for {method} {path}")  # => co-03: caller should map this to a 404
 
 
def get_item(params: dict[str, str]) -> str:  # => a tiny handler -- holds no routing logic itself
    return f"item {params['id']}"
 
 
router = Router()
router.add("GET", "/items/{id}", get_item)  # => co-07: registers the pattern once
 
result = router.dispatch("GET", "/items/42")  # => co-12: "42" is extracted and bound to params["id"]
print(result)  # => Output: item 42
 
assert result == "item 42"
try:
    router.dispatch("GET", "/items/42/extra")  # => co-03: segment count mismatch -- no route matches
    raised = False
except LookupError:
    raised = True
print(raised)  # => Output: True
assert raised is True
print("kata-09 OK")

item 42
True
kata-09 OK

import json
from typing import cast
 
 
def parse_json_body(raw_body: str) -> tuple[dict[str, object] | None, str | None]:
    # => co-09/co-13: (de)serialize the raw request body into a typed Python object for the handler
    try:
        parsed = json.loads(raw_body)  # => co-09: JSON text -> Python value (json.loads returns Any)
    except json.JSONDecodeError as exc:  # => co-13: malformed body -- must NOT crash the handler
        return None, f"invalid JSON: {exc.msg}"  # => a caller-facing reason, not a raw traceback (co-11)
    if not isinstance(parsed, dict):  # => co-13: the handler expects an OBJECT body, not a bare list/number
        return None, "request body must be a JSON object"
    return cast(dict[str, object], parsed), None  # => JSON objects always have str keys -- safe to narrow
 
 
good_raw = '{"title": "Ship report", "done": false}'  # => a well-formed JSON object body
malformed_raw = '{"title": "Ship report",}'  # => a trailing comma -- invalid JSON syntax
wrong_shape_raw = "[1, 2, 3]"  # => valid JSON, but not an OBJECT -- still not usable as a request body
 
good_parsed, good_error = parse_json_body(good_raw)
malformed_parsed, malformed_error = parse_json_body(malformed_raw)
wrong_shape_parsed, wrong_shape_error = parse_json_body(wrong_shape_raw)
 
print(good_parsed, good_error)  # => Output: {'title': 'Ship report', 'done': False} None
print(malformed_parsed, malformed_error is not None)  # => Output: None True
print(wrong_shape_parsed, wrong_shape_error)  # => Output: None request body must be a JSON object
 
assert good_parsed == {"title": "Ship report", "done": False} and good_error is None
assert malformed_parsed is None and malformed_error is not None  # => rejected, not silently truncated
assert wrong_shape_parsed is None and wrong_shape_error == "request body must be a JSON object"
print("kata-10 OK")

{'title': 'Ship report', 'done': False} None
None True
None request body must be a JSON object
kata-10 OK

from typing import TypedDict
 
 
class Task(TypedDict):  # => co-14: the repo returns a TYPED shape, not a loose dict-of-anything
    id: int
    title: str
    done: bool
 
 
class TaskRepository:  # => co-14/co-24: the ONLY place that touches the "store" -- a mocked in-memory table
    def __init__(self) -> None:
        self._rows: dict[int, Task] = {}  # => stands in for a real DB table for this pure kata
        self._next_id = 1  # => stands in for an auto-increment primary key
 
    def create(self, title: str) -> Task:  # => co-14: INSERT-equivalent
        task: Task = {"id": self._next_id, "title": title, "done": False}
        self._rows[self._next_id] = task
        self._next_id += 1
        return task
 
    def get(self, task_id: int) -> Task | None:  # => co-14: SELECT-by-id-equivalent
        return self._rows.get(task_id)  # => None if absent -- the CALLER decides how to map that to a 404
 
    def update_done(self, task_id: int, done: bool) -> Task | None:  # => co-14: UPDATE-equivalent
        task = self._rows.get(task_id)
        if task is None:  # => co-14: nothing to update -- signal absence, don't raise a generic error
            return None
        task["done"] = done  # => in a real repo this would be a parameterized UPDATE ... WHERE id = ?
        return task
 
    def delete(self, task_id: int) -> bool:  # => co-14: DELETE-equivalent
        return self._rows.pop(task_id, None) is not None  # => True only if a row actually existed
 
 
def complete_task_handler(repo: TaskRepository, task_id: int) -> dict[str, object]:
    # => co-24: the HANDLER holds zero persistence logic -- it only calls the repo and shapes a response
    task = repo.update_done(task_id, True)  # => all the "how do we store this" detail lives in the repo
    if task is None:
        return {"status": 404, "body": {"error": "task not found"}}  # => co-11: a structured 404
    return {"status": 200, "body": task}
 
 
repo = TaskRepository()
created = repo.create("Write the report")  # => co-14: id=1 assigned by the repo, not the handler
print(created)  # => Output: {'id': 1, 'title': 'Write the report', 'done': False}
assert created == {"id": 1, "title": "Write the report", "done": False}
 
response = complete_task_handler(repo, created["id"])  # => the handler never sees a single line of SQL
missing_response = complete_task_handler(repo, 999)  # => a task id that was never created
 
print(response)  # => Output: {'status': 200, 'body': {'id': 1, 'title': 'Write the report', 'done': True}}
print(missing_response)  # => Output: {'status': 404, 'body': {'error': 'task not found'}}
 
assert response["status"] == 200
body = response["body"]
assert isinstance(body, dict) and body["done"] is True  # => the repo's update actually persisted
assert missing_response["status"] == 404  # => co-14/co-11: a clean 404, not a KeyError leaking upward
print("kata-11 OK")

{'id': 1, 'title': 'Write the report', 'done': False}
{'status': 200, 'body': {'id': 1, 'title': 'Write the report', 'done': True}}
{'status': 404, 'body': {'error': 'task not found'}}
kata-11 OK

class Worker:  # => co-05: models one server PROCESS -- its own local memory, no memory shared with peers
    def __init__(self, name: str) -> None:
        self.name = name
        self._local_cache: dict[str, int] = {}  # => in-process state -- this is what statelessness FORBIDS relying on
 
    def handle_request(self, shared_db: dict[str, int], key: str, increment: int) -> int:
        # => co-05: the ONLY state this handler is allowed to depend on is the SHARED store, not local memory
        self._local_cache[key] = self._local_cache.get(key, 0) + 1  # => a LOCAL counter -- diverges per worker
        shared_db[key] = shared_db.get(key, 0) + increment  # => co-05: durable state lives in ONE shared place
        return shared_db[key]  # => every worker reads/writes the SAME shared value, regardless of which handled it
 
    def local_count(self, key: str) -> int:  # => a PUBLIC accessor -- for this kata's inspection only
        return self._local_cache.get(key, 0)
 
 
shared_db: dict[str, int] = {}  # => co-05: stands in for "the database" -- the one place real state lives
worker_a = Worker("worker-a")  # => co-05: two independent workers, as if behind a load balancer
worker_b = Worker("worker-b")
 
result1 = worker_a.handle_request(shared_db, "views", 1)  # => request 1 happens to land on worker A
result2 = worker_b.handle_request(shared_db, "views", 1)  # => request 2 happens to land on worker B
result3 = worker_a.handle_request(shared_db, "views", 1)  # => request 3 lands back on worker A
 
print(result1, result2, result3)  # => Output: 1 2 3
print(worker_a.local_count("views"), worker_b.local_count("views"))  # => Output: 2 1
 
# => co-05: each worker's LOCAL count only reflects requests IT happened to handle -- they diverge
assert worker_a.local_count("views") == 2 and worker_b.local_count("views") == 1
# => but the SHARED db is consistent regardless of which worker handled which request -- that's the point
assert shared_db["views"] == 3 == result3
print("kata-12 OK")

1 2 3
2 1
kata-12 OK

from collections.abc import Callable
from typing import TypedDict
 
 
class Response(TypedDict):  # => co-04: a typed shape, so `headers` is always known to be dict[str, str]
    status: int
    body: str
    headers: dict[str, str]
 
 
Handler = Callable[[dict[str, str]], Response]  # => co-08: a plain handler -- knows nothing about middleware
Middleware = Callable[[Handler], Handler]  # => co-16: wraps a handler, returns a NEW handler
 
 
def request_id_middleware(counter: list[int]) -> Middleware:  # => co-16/co-04: adds X-Request-Id
    def wrap(handler: Handler) -> Handler:
        def wrapped(headers: dict[str, str]) -> Response:
            counter[0] += 1  # => a mocked id generator -- a real one might use uuid4()
            response = handler(headers)  # => call the INNER handler first
            response["headers"]["X-Request-Id"] = str(counter[0])  # => co-04: mutate the typed headers dict
            return response
 
        return wrapped
 
    return wrap
 
 
def timing_middleware(clock: Callable[[], float]) -> Middleware:  # => co-16: adds X-Process-Time
    def wrap(handler: Handler) -> Handler:
        def wrapped(headers: dict[str, str]) -> Response:
            start = clock()  # => mocked clock -- avoids real-time flakiness in this kata
            response = handler(headers)
            elapsed = clock() - start  # => wall-clock cost of the WRAPPED handler only
            response["headers"]["X-Process-Time"] = f"{elapsed:.3f}"  # => co-04/co-16
            return response
 
        return wrapped
 
    return wrap
 
 
def base_handler(headers: dict[str, str]) -> Response:  # => co-08: the actual endpoint logic, middleware-free
    return {"status": 200, "body": "ok", "headers": {}}
 
 
ticks = iter([0.0, 0.25])  # => mocked clock readings: start=0.0, end=0.25 -- deterministic, no real sleep
request_counter = [0]
 
wrapped_handler = timing_middleware(lambda: next(ticks))(  # => co-16: compose middlewares around base_handler
    request_id_middleware(request_counter)(base_handler)
)
response = wrapped_handler({})
 
print(response)  # => Output: {'status': 200, 'body': 'ok', 'headers': {'X-Request-Id': '1', 'X-Process-Time': '0.250'}}
 
headers = response["headers"]
assert headers["X-Request-Id"] == "1"  # => co-16/co-04: every response now carries a traceable id
assert headers["X-Process-Time"] == "0.250"  # => co-16: cross-cutting behavior added WITHOUT touching base_handler
print("kata-13 OK")

{'status': 200, 'body': 'ok', 'headers': {'X-Request-Id': '1', 'X-Process-Time': '0.250'}}
kata-13 OK

from collections.abc import Callable
from typing import TypedDict
 
 
class Task(TypedDict):
    id: int
    status: str
    priority: str
 
 
FILTER_KEYS: dict[str, Callable[[Task], str]] = {  # => co-20: an allowlist -- never eval() a raw field name
    "status": lambda t: t["status"],
    "priority": lambda t: t["priority"],
}
SORT_KEYS: dict[str, Callable[[Task], str]] = FILTER_KEYS  # => same typed accessors double as sort keys
 
 
def filter_and_sort(tasks: list[Task], filters: dict[str, str], sort_key: str | None) -> list[Task]:  # => co-20: query-param-like filters narrow the list, an optional sort orders it
    result = [  # => co-20: AND semantics -- a row must match EVERY supplied filter, not just one
        task for task in tasks if all(FILTER_KEYS[field](task) == value for field, value in filters.items())
    ]
    if sort_key is not None:  # => co-20: sort is optional -- absent means "keep insertion order"
        result = sorted(result, key=SORT_KEYS[sort_key])  # => Timsort: stable, O(n log n)
    return result
 
 
tasks: list[Task] = [
    {"id": 1, "status": "open", "priority": "high"},
    {"id": 2, "status": "done", "priority": "low"},
    {"id": 3, "status": "done", "priority": "high"},
    {"id": 4, "status": "open", "priority": "low"},
]
 
done_only = filter_and_sort(tasks, {"status": "done"}, None)  # => single filter, no sort
done_high = filter_and_sort(tasks, {"status": "done", "priority": "high"}, None)  # => co-20: two filters, AND'd
sorted_by_priority = filter_and_sort(tasks, {}, "priority")  # => no filter at all, just a sort
 
print([t["id"] for t in done_only])  # => Output: [2, 3]
print([t["id"] for t in done_high])  # => Output: [3]
print([t["id"] for t in sorted_by_priority])  # => Output: [1, 3, 2, 4] -- alphabetical: high, high, low, low
 
assert [t["id"] for t in done_only] == [2, 3]
assert [t["id"] for t in done_high] == [3]  # => id 2 is done but low priority -- excluded by the AND
assert [t["id"] for t in sorted_by_priority] == [1, 3, 2, 4]
print("kata-14 OK")

[2, 3]
[3]
[1, 3, 2, 4]
kata-14 OK

def validate_constraints(body: dict[str, object]) -> list[str]:  # => co-10: mimics Pydantic-style field rules
    errors: list[str] = []  # => co-11: collect ALL violations, not just the first -- one 422 lists them all
    title = body.get("title")
    if not isinstance(title, str) or len(title) < 3:  # => co-10: min_length=3 equivalent
        errors.append("title: must be a string with min_length 3")
    quantity = body.get("quantity")
    if not isinstance(quantity, int) or quantity <= 0:  # => co-10: gt=0 equivalent
        errors.append("quantity: must be an integer greater than 0")
    return errors  # => empty list means the body satisfies every constraint
 
 
valid_body: dict[str, object] = {"title": "Ship it", "quantity": 3}
short_title_body: dict[str, object] = {"title": "Hi", "quantity": 3}  # => violates min_length
bad_quantity_body: dict[str, object] = {"title": "Ship it", "quantity": 0}  # => violates gt=0
both_bad_body: dict[str, object] = {"title": "Hi", "quantity": -1}  # => violates BOTH constraints
 
print(validate_constraints(valid_body))  # => Output: []
print(validate_constraints(short_title_body))  # => Output: ['title: must be a string with min_length 3']
print(validate_constraints(bad_quantity_body))  # => Output: ['quantity: must be an integer greater than 0']
print(validate_constraints(both_bad_body))  # => Output: ['title: must be a string with min_length 3', 'quantity: must be an integer greater than 0']
 
assert validate_constraints(valid_body) == []  # => co-10: nothing to reject -- the body is valid
assert len(validate_constraints(short_title_body)) == 1
assert len(validate_constraints(both_bad_body)) == 2  # => co-11: BOTH violations reported in one 422, not just the first
print("kata-15 OK")

[]
['title: must be a string with min_length 3']
['quantity: must be an integer greater than 0']
['title: must be a string with min_length 3', 'quantity: must be an integer greater than 0']
kata-15 OK

from typing import TypedDict
 
 
class TaskV1(TypedDict):  # => co-15: the SHAPE before the migration -- what existing rows already look like
    id: int
    title: str
 
 
class TaskV2(TypedDict):  # => co-15: the SHAPE after the migration -- one new nullable-with-default column
    id: int
    title: str
    archived: bool  # => the new column -- an ADDITIVE change, nothing removed or renamed
 
 
def migrate_add_archived(rows: list[TaskV1]) -> list[TaskV2]:  # => co-15: ALTER TABLE ... ADD COLUMN + backfill
    return [
        {"id": row["id"], "title": row["title"], "archived": False}  # => co-15: backfill every existing row
        for row in rows  # => a real migration would run one UPDATE, not a Python loop -- same idea though
    ]
 
 
existing_rows: list[TaskV1] = [  # => co-15: rows that existed BEFORE the migration ran
    {"id": 1, "title": "Write the report"},
    {"id": 2, "title": "Ship the release"},
]
 
migrated_rows = migrate_add_archived(existing_rows)
print(migrated_rows)  # => Output: [{'id': 1, 'title': 'Write the report', 'archived': False}, {'id': 2, 'title': 'Ship the release', 'archived': False}]
 
# => co-15: EVERY pre-existing row gained the new column with a safe default -- nothing became invalid
assert all(row["archived"] is False for row in migrated_rows)
assert [row["id"] for row in migrated_rows] == [1, 2]  # => co-15: identity and order fully preserved
assert len(migrated_rows) == len(existing_rows)  # => co-15: additive means no row is dropped or merged
print("kata-16 OK")

[{'id': 1, 'title': 'Write the report', 'archived': False}, {'id': 2, 'title': 'Ship the release', 'archived': False}]
kata-16 OK

def accepts_json_body(content_type: str | None) -> tuple[bool, int]:  # => returns (should_parse, status)
    # => co-21: mirrors FastAPI's strict_content_type=True default (since 0.132.0) -- reject non-JSON bodies
    if content_type is None:  # => no Content-Type header sent at all
        return False, 422  # => co-21: still 422 -- FastAPI parses this as "body didn't match the model"
    media_type = content_type.split(";")[0].strip().lower()  # => strip a "; charset=utf-8" suffix if present
    if media_type == "application/json":  # => co-21: the only media type this endpoint accepts as JSON
        return True, 200
    return False, 422  # => co-21: NOT a 415 -- FastAPI's default has no dedicated 415 for this case
 
 
json_ct = "application/json"  # => the well-formed, expected case
json_ct_with_charset = "application/json; charset=utf-8"  # => co-21: a charset suffix must not break matching
plain_text_ct = "text/plain"  # => co-21: JSON body sent with the wrong declared Content-Type
missing_ct = None  # => co-21: no Content-Type header at all
 
for label, ct in [
    ("json", json_ct),
    ("json+charset", json_ct_with_charset),
    ("plain-text", plain_text_ct),
    ("missing", missing_ct),
]:
    accepted, status = accepts_json_body(ct)
    print(label, accepted, status)
    # => Output rows: json True 200 / json+charset True 200 / plain-text False 422 / missing False 422
 
assert accepts_json_body(json_ct) == (True, 200)
assert accepts_json_body(json_ct_with_charset) == (True, 200)  # => co-21: charset suffix tolerated
assert accepts_json_body(plain_text_ct) == (False, 422)  # => co-21: rejected -- a framework default, not hand-written
assert accepts_json_body(missing_ct) == (False, 422)
print("kata-17 OK")

json True 200
json+charset True 200
plain-text False 422
missing False 422
kata-17 OK

from collections.abc import Callable
 
 
class Connection:  # => stands in for a real DB connection object
    def __init__(self, name: str) -> None:
        self.name = name
        self.closed = False
 
    def close(self) -> None:
        self.closed = True
 
 
def get_connection(pool: list[Connection]) -> Connection:  # => co-23: a FastAPI-Depends-style PROVIDER
    return pool.pop()  # => hands out ONE connection per call, taking it out of the shared pool
 
 
def run_with_dependency(  # => co-23: mimics what FastAPI does with `Depends(get_connection)` per request
    provider: Callable[[], Connection], handler: Callable[[Connection], str]
) -> str:
    connection = provider()  # => co-23: the FRAMEWORK resolves the dependency -- the handler never calls it
    try:
        return handler(connection)  # => co-23: the handler just RECEIVES a ready-to-use resource
    finally:
        connection.close()  # => co-23: the framework is responsible for teardown too, not the handler
 
 
def list_tasks_handler(connection: Connection) -> str:  # => co-08: a thin handler -- no wiring logic itself
    return f"tasks from {connection.name}"
 
 
pool = [Connection("conn-a"), Connection("conn-b")]  # => co-23: a mocked connection pool
provider = lambda: get_connection(pool)  # => co-23: bound once, reused per simulated "request"
 
result = run_with_dependency(provider, list_tasks_handler)  # => request 1: injected with conn-b (last popped)
print(result)  # => Output: tasks from conn-b
print(len(pool))  # => Output: 1 -- whatever's left in the pool after one injection
 
assert result == "tasks from conn-b"
assert len(pool) == 1  # => co-23: exactly one connection was handed out and consumed
print("kata-18 OK")

tasks from conn-b
1
kata-18 OK

from collections.abc import Callable
from typing import TypedDict
 
 
class Task(TypedDict):
    id: int
    status: str
    priority: int
 
 
class Page(TypedDict):  # => co-19: a typed pagination envelope -- items, total, and a next-page cursor
    items: list[Task]
    total: int
    next_offset: int | None
 
 
# => co-20: a small allowlist of sortable fields, each with an explicit typed accessor (never eval()'d)
SORT_KEYS: dict[str, Callable[[Task], int]] = {"priority": lambda t: t["priority"]}
 
 
def list_tasks(  # => co-19/co-20: pagination + filter + sort composed in ONE query-shaping function
    tasks: list[Task],
    status: str | None,
    sort_by: str | None,
    limit: int,
    offset: int,
) -> Page:
    filtered = [t for t in tasks if status is None or t["status"] == status]  # => co-20: filter step first
    if sort_by is not None:  # => co-20: sort step second, over the ALREADY-filtered set
        filtered = sorted(filtered, key=SORT_KEYS[sort_by])
    page = filtered[offset : offset + limit]  # => co-19: pagination step LAST, over the filtered+sorted set
    return {  # => co-19: pagination metadata alongside the page itself
        "items": page,
        "total": len(filtered),  # => co-19: total reflects the FILTERED count, not the whole table
        "next_offset": offset + limit if offset + limit < len(filtered) else None,
    }
 
 
tasks: list[Task] = [
    {"id": 1, "status": "open", "priority": 3},
    {"id": 2, "status": "open", "priority": 1},
    {"id": 3, "status": "done", "priority": 2},
    {"id": 4, "status": "open", "priority": 2},
]
 
page1 = list_tasks(tasks, status="open", sort_by="priority", limit=2, offset=0)  # => co-19/co-20: all 3 combine
page2 = list_tasks(tasks, status="open", sort_by="priority", limit=2, offset=2)  # => the SECOND page
 
print([t["id"] for t in page1["items"]], page1["total"], page1["next_offset"])  # => Output: [2, 4] 3 2
print([t["id"] for t in page2["items"]], page2["total"], page2["next_offset"])  # => Output: [1] 3 None
 
assert [t["id"] for t in page1["items"]] == [2, 4]  # => sorted by priority (1, 2) among "open" tasks
assert page1["total"] == 3  # => co-19: 3 total open tasks, even though only 2 are on this page
assert page1["next_offset"] == 2  # => co-19: a next page exists
assert page2["next_offset"] is None  # => co-19: this page reaches the end of the filtered set
print("kata-19 OK")

[2, 4] 3 2
[1] 3 None
kata-19 OK