Build an API server from scratch

Requests and responses as objects

Nobody wants to pick through raw bytes in every function. A kitchen works from a tidy order slip, and so should your code. So parse the bytes once, and pass around small objects. Dataclasses, from Phase 7, fit well:

from dataclasses import dataclass, field
import json

@dataclass
class Request:
    method: str
    path: str
    query: dict = field(default_factory=dict)
    headers: dict = field(default_factory=dict)
    body: bytes = b""

    def json(self):
        return json.loads(self.body or b"null")

@dataclass
class Response:
    status: int
    body: object = None
    headers: dict = field(default_factory=dict)

def hello(req):
    name = req.query.get("name", "world")
    return Response(200, {"message": f"Hello, {name}!"})

print(hello(Request("GET", "/hello", query={"name": "Ada"})))

A handler is a function from a Request to a Response: one dish on the menu. The rest of the server exists to pick the right handler and turn its Response into bytes.

Routing

Think of an office switchboard. A call comes in asking for sales, and the operator looks up the extension. The router does that job. Each route is a row in its table: a method, a path pattern and a handler.

A pattern can hold path parameters in braces, which work like blanks on a form. /campaigns/{id} matches /campaigns/cmp_103 and captures id = "cmp_103".

Matching works one segment at a time. A segment is the text between slashes, so /campaigns/cmp_103 has two. Compare them in pairs: campaigns must match exactly, and {id} accepts anything that isn't empty. Different numbers of segments can never match.

def match_pattern(pattern, path):
    p_parts = pattern.strip("/").split("/")
    parts = path.strip("/").split("/")
    if len(p_parts) != len(parts):
        return None
    params = {}
    for p, actual in zip(p_parts, parts):
        if p.startswith("{") and p.endswith("}"):
            if not actual:
                return None
            params[p[1:-1]] = actual
        elif p != actual:
            return None
    return params

print(match_pattern("/campaigns/{id}", "/campaigns/cmp_103"))
print(match_pattern("/campaigns/{id}", "/campaigns"))
print(match_pattern("/users/{uid}/orders/{oid}", "/users/7/orders/o9"))

404 or 405?

Say you walk down a street looking for a bakery. There are two different ways to be turned away. If there's no bakery on that street, that's one problem. If you find it and ask them to fix your car, that's another.

• 404 Not Found is the missing bakery. No route's pattern matches the path at all. GET /campaings (a typo) is an example.
• 405 Method Not Allowed is asking the wrong thing of the right place. The path matches a pattern, but not with this method. DELETE /campaigns when only GET and POST exist there is an example.

A 405 must include an Allow header listing the methods that do work, such as Allow: GET, POST. That's how the client learns what it can do instead. Getting this right is a small sign of a carefully built API.

routes = {"/items": ["GET"]}
path, method = "/items", "DELETE"
if path not in routes: print(404)
elif method not in routes[path]: print(405)
else: print(200)

import json

CAMPAIGNS = {"cmp_1": {"id": "cmp_1", "name": "Spring sale", "budget": 500}}
ROUTES = []

def route(method, pattern):
    def register(handler):
        ROUTES.append((method, pattern, handler))
        return handler
    return register

def match_pattern(pattern, path):
    p_parts, parts = pattern.strip("/").split("/"), path.strip("/").split("/")
    if len(p_parts) != len(parts):
        return None
    params = {}
    for p, a in zip(p_parts, parts):
        if p.startswith("{"):
            params[p[1:-1]] = a
        elif p != a:
            return None
    return params

@route("GET", "/campaigns")
def list_campaigns(params, body):
    return 200, {"data": list(CAMPAIGNS.values())}

@route("GET", "/campaigns/{id}")
def get_campaign(params, body):
    c = CAMPAIGNS.get(params["id"])
    return (200, c) if c else (404, {"error": f"no campaign {params['id']}"})

@route("POST", "/campaigns")
def create_campaign(params, body):
    new_id = f"cmp_{len(CAMPAIGNS) + 1}"
    CAMPAIGNS[new_id] = {"id": new_id, **body}
    return 201, CAMPAIGNS[new_id]

def handle(method, path, body=None):
    allowed = []
    for m, pattern, handler in ROUTES:
        params = match_pattern(pattern, path)
        if params is None:
            continue
        if m == method:
            return handler(params, body)
        allowed.append(m)
    if allowed:
        return 405, {"error": "method not allowed", "allow": sorted(allowed)}
    return 404, {"error": "not found"}

print(handle("POST", "/campaigns", {"name": "Autumn", "budget": 300}))
print(handle("GET", "/campaigns/cmp_2"))
print(handle("DELETE", "/campaigns"))
print(handle("GET", "/campaings"))

WSGI and ASGI: the standard plug

Any lamp works in any wall socket, because everyone agreed on the shape of the plug. Python web servers and frameworks made the same kind of agreement about step 3. So any server can run any framework.

• WSGI (Web Server Gateway Interface) is the classic plug. An app is a function app(environ, start_response). environ is a dict describing the request, and the app calls start_response(status, headers) and returns the body. Flask and Django speak WSGI, and servers such as Gunicorn run them.
• ASGI is the newer one, for async code. Apps are async functions, so one process can hold thousands of open connections, including WebSockets and streams. It's like a waiter who takes one table's order and moves on, without standing there until the food is ready. FastAPI and Starlette speak ASGI, and Uvicorn runs them.

from wsgiref.util import setup_testing_defaults

def app(environ, start_response):
    body = f"You asked for {environ['PATH_INFO']}".encode()
    start_response("200 OK", [("Content-Type", "text/plain"), ("Content-Length", str(len(body)))])
    return [body]

environ = {"PATH_INFO": "/campaigns", "REQUEST_METHOD": "GET"}
setup_testing_defaults(environ)          # fill in the rest of a valid request
captured = {}
result = app(environ, lambda status, headers: captured.update(status=status, headers=headers))
print(captured["status"], b"".join(result))

That call is all a WSGI server does with your app, once per request. Because it's a plain function call, frameworks provide test clients that call the app directly in exactly this way.

Why build one by hand at all

You'll use a framework in real work. But having written this, you know a framework isn't magic. It's a loop, a parser, a route table and a function call. So when something strange happens, such as a header that goes missing or a request that hangs, you know where to look. HTTP doesn't require any framework, and that knowledge outlasts any one of them.

Getting the small things right

Answer 404 and 405 correctly. Getting it right tells a developer whether they've misspelled the URL or used the wrong verb, which saves them a long hunt.

Check your path patterns. A pattern like /items/<id> shouldn't match /items/. If it does, your handler receives an empty id, and you might end up listing or deleting the wrong thing. Test that case on purpose.

Test without a network. Your app is only a function that takes a request and returns a response. You can taste a dish without opening the restaurant, so call it directly:

def app(method, path):
    if path == "/items" and method == "GET":
        return 200, ["tent"]
    if path == "/items":
        return 405, "use GET"
    return 404, "not found"

print(app("GET", "/items"))
print(app("DELETE", "/items"))
print(app("GET", "/nope"))

Tests like these run thousands of times faster than starting a real server and sending requests. They can't fail because a port is busy, either. Keep a few real end-to-end tests, and do the rest this way.

What a real server adds

Your tiny server is one waiter, serving one table at a time. Production servers like Gunicorn and Uvicorn are the rest of the staff. They add many workers, time limits, limits on body size and protection from slow clients. You don't write those parts yourself. You plug your app into one, through the WSGI or ASGI interface you met in this lesson.