S055 · HTTP Fundamentals — Verbs, Status Codes, Headers, Caching

"""
http_demo.py — A minimal HTTP server that demonstrates verbs, status codes,
ETags, cache directives, and JSON bodies.
 
Run:
    python http_demo.py
And in another terminal:
    curl -v http://127.0.0.1:8080/users/42
    curl -v -X POST -d '{"name":"alice"}' -H 'Content-Type: application/json' http://127.0.0.1:8080/users
    curl -v http://127.0.0.1:8080/users/999
    curl -v -H 'If-None-Match: "v1-42"' http://127.0.0.1:8080/users/42
"""
from __future__ import annotations
 
import hashlib
import http.server
import json
from datetime import datetime, timezone
 
USERS: dict[int, dict] = {42: {"id": 42, "name": "alice", "email": "a@x.io"}}
NEXT_ID = 100
 
 
def make_etag(payload: dict) -> str:
    raw = json.dumps(payload, sort_keys=True).encode()
    digest = hashlib.md5(raw).hexdigest()[:8]
    return f'"v1-{digest}"'
 
 
class Handler(http.server.BaseHTTPRequestHandler):
    # Silence default logging
    def log_message(self, fmt: str, *args) -> None:
        print(f"[{datetime.now(timezone.utc).isoformat()}] " + fmt % args)
 
    # ---------- helpers ----------
    def _send_json(self, status: int, payload: dict, extra_headers: dict | None = None) -> None:
        body = json.dumps(payload).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        for k, v in (extra_headers or {}).items():
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(body)
 
    def _parse_id(self) -> int | None:
        parts = self.path.strip("/").split("/")
        if len(parts) != 2 or parts[0] != "users":
            return None
        try:
            return int(parts[1])
        except ValueError:
            return None
 
    # ---------- verbs ----------
    def do_GET(self) -> None:
        uid = self._parse_id()
        if uid is None:
            self._send_json(400, {"error": "bad path"})
            return
        user = USERS.get(uid)
        if not user:
            self._send_json(404, {"error": "not found"})
            return
        etag = make_etag(user)
        if_none_match = self.headers.get("If-None-Match")
        if if_none_match == etag:
            self.send_response(304)
            self.send_header("ETag", etag)
            self.end_headers()
            return
        self._send_json(
            200, user,
            extra_headers={
                "ETag": etag,
                "Cache-Control": "public, max-age=60, s-maxage=3600, stale-while-revalidate=86400",
            },
        )
 
    def do_POST(self) -> None:
        global NEXT_ID
        if self.path != "/users":
            self._send_json(404, {"error": "no route"})
            return
        length = int(self.headers.get("Content-Length", "0"))
        raw = self.rfile.read(length)
        try:
            payload = json.loads(raw)
        except json.JSONDecodeError:
            self._send_json(400, {"error": "invalid json"})
            return
        if "name" not in payload:
            self._send_json(422, {"error": "missing field: name"})
            return
        user = {"id": NEXT_ID, **payload}
        USERS[NEXT_ID] = user
        NEXT_ID += 1
        self._send_json(
            201, user,
            extra_headers={"Location": f"/users/{user['id']}"},
        )
 
    def do_PUT(self) -> None:
        uid = self._parse_id()
        if uid is None:
            self._send_json(400, {"error": "bad path"})
            return
        length = int(self.headers.get("Content-Length", "0"))
        payload = json.loads(self.rfile.read(length))
        payload["id"] = uid
        USERS[uid] = payload
        self._send_json(200, payload)
 
    def do_DELETE(self) -> None:
        uid = self._parse_id()
        if uid is None or uid not in USERS:
            self._send_json(404, {"error": "not found"})
            return
        del USERS[uid]
        self.send_response(204)
        self.end_headers()
 
 
def main() -> None:
    server = http.server.HTTPServer(("127.0.0.1", 8080), Handler)
    print("Serving on http://127.0.0.1:8080")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nBye.")
 
 
if __name__ == "__main__":
    main()

python http_demo.py    # in one terminal
 
# 200 with a body + ETag + Cache-Control
curl -v http://127.0.0.1:8080/users/42
 
# 304 Not Modified (conditional GET) — take the ETag from the previous response
curl -v -H 'If-None-Match: "v1-<hash>"' http://127.0.0.1:8080/users/42
 
# 201 Created with Location header
curl -v -X POST -d '{"name":"bob"}' -H 'Content-Type: application/json' http://127.0.0.1:8080/users
 
# 422 Unprocessable Entity (validation)
curl -v -X POST -d '{}' -H 'Content-Type: application/json' http://127.0.0.1:8080/users
 
# 204 No Content (successful delete)
curl -v -X DELETE http://127.0.0.1:8080/users/42