Serve the contract

A server and an application

The previous lesson defined Relay's contract: POST /jobs accepts a job and GET /jobs/{id} returns its status.

Now something must listen on a network port, read each request, and send a response. Python splits that work in two. A server accepts connections and parses the bytes of HTTP; an application decides what each request means.

WSGI, the Web Server Gateway Interface defined in PEP 3333, is the agreement between them. The application is a function with two arguments: environ, a dictionary that describes the request, and start_response, a function it calls with the status and the headers. It returns the body as an iterable of bytes, such as a list with one bytes object.

Flask and Django applications are WSGI applications too, so this lesson shows what those frameworks do for you.

import json

def app(environ, start_response):
    body = json.dumps({"status": "ok"}).encode("utf-8")
    start_response("200 OK", [("Content-Type", "application/json")])
    return [body]

Route, read, and answer

environ["REQUEST_METHOD"] holds the method and environ["PATH_INFO"] the path, so routing is a decision on those two values.

For POST /jobs, read the body from environ["wsgi.input"], exactly as many bytes as CONTENT_LENGTH says. That variable may be missing or empty, and the application should never read past it. Decode the bytes as JSON and check the shape before creating work.

Each outcome gets its own status: 202 Accepted with the new job's id, 400 Bad Request for a body that is not valid JSON or lacks a field, 404 Not Found for an unknown job, and 405 Method Not Allowed for a known path with the wrong method, which must list the allowed methods in an Allow header. The status is text such as "404 Not Found", headers are pairs of strings, and the body is bytes.

def app(environ, start_response):
    method = environ["REQUEST_METHOD"]
    path = environ.get("PATH_INFO", "")
    if path == "/jobs" and method == "POST":
        return create_job(environ, start_response)
    if path.startswith("/jobs/") and method == "GET":
        return show_job(path.removeprefix("/jobs/"), start_response)
    return respond(start_response, "404 Not Found", {"error": "not_found"})

Run it, then test it without a network

The standard library's wsgiref.simple_server serves the application on your own computer: make_server("", 8000, app).serve_forever() answers on port 8000, and a client such as curl can send it requests.

Its documentation is clear that wsgiref is a reference implementation, not recommended for production. A deployment runs the same application under a server built for that work, such as Gunicorn or Waitress, which handle many connections, timeouts, and restarts. Because the application is only a function, a test can call it directly: build a small environ dictionary, pass a start_response that records the status and headers, and read the returned bytes.

For now Relay keeps its jobs in a dictionary, so a 202 promises only memory; the next lessons move them to SQLite.

import io

def request(method, path, body=b""):
    seen = {}
    def start_response(status, headers, exc_info=None):
        seen["status"] = status
    environ = {"REQUEST_METHOD": method, "PATH_INFO": path,
               "CONTENT_LENGTH": str(len(body)), "wsgi.input": io.BytesIO(body)}
    chunks = app(environ, start_response)
    return seen["status"], b"".join(chunks)