Capstone: a full-stack course-booking application

# seatwise.py: the application and its tests in one file, run in-process (no network needed).
import hashlib, hmac, html, io, json, secrets, sqlite3, base64, re
from urllib.parse import parse_qs

SCHEMA = """
CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, email TEXT UNIQUE NOT NULL, role TEXT NOT NULL DEFAULT 'student');
CREATE TABLE IF NOT EXISTS sections (id TEXT PRIMARY KEY, course TEXT NOT NULL, title TEXT NOT NULL,
                                     capacity INTEGER NOT NULL CHECK (capacity > 0));
CREATE TABLE IF NOT EXISTS registrations (id INTEGER PRIMARY KEY, section_id TEXT NOT NULL REFERENCES sections(id),
                                       user_id INTEGER NOT NULL REFERENCES users(id), UNIQUE (section_id, user_id));
"""

class CourseBook:
    """Seatwise: a small course-section booking service (WSGI)."""
    def __init__(self, db_path=":memory:", secret=None):
        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self.db.execute("PRAGMA foreign_keys = ON"); self.db.executescript(SCHEMA)
        self.secret = secret or secrets.token_bytes(32)

    # --- sessions ---------------------------------------------------------
    def sign(self, data):
        raw = base64.urlsafe_b64encode(json.dumps(data).encode()).decode()
        return raw + "." + hmac.new(self.secret, raw.encode(), hashlib.sha256).hexdigest()
    def session(self, environ):
        m = re.search(r"(?:^|;\s*)__Host-sid=([^;]+)", environ.get("HTTP_COOKIE", ""))
        if not m: return {}
        raw, _, mac = m.group(1).rpartition(".")
        good = hmac.new(self.secret, raw.encode(), hashlib.sha256).hexdigest()
        return json.loads(base64.urlsafe_b64decode(raw)) if hmac.compare_digest(mac, good) else {}

    # --- domain -----------------------------------------------------------
    def seats_left(self, sid):
        cap, = self.db.execute("SELECT capacity FROM sections WHERE id = ?", (sid,)).fetchone()
        n, = self.db.execute("SELECT count(*) FROM registrations WHERE section_id = ?", (sid,)).fetchone()
        return cap - n
    def register(self, sid, uid):
        with self.db:                                         # BEGIN IMMEDIATE serializes writers in SQLite
            self.db.execute("BEGIN IMMEDIATE") if not self.db.in_transaction else None
            if self.db.execute("SELECT 1 FROM sections WHERE id = ?", (sid,)).fetchone() is None:
                return 404, {"title": "Not Found", "detail": "no such section"}
            if self.seats_left(sid) <= 0:
                return 409, {"title": "Conflict", "detail": "section is full"}
            try:
                self.db.execute("INSERT INTO registrations (section_id, user_id) VALUES (?, ?)", (sid, uid))
            except sqlite3.IntegrityError:
                return 409, {"title": "Conflict", "detail": "already registered"}
        return 201, {"section": sid, "seats_left": self.seats_left(sid)}

    # --- HTTP -------------------------------------------------------------
    def __call__(self, environ, start_response):
        method, path = environ["REQUEST_METHOD"], environ["PATH_INFO"]
        sess = self.session(environ)
        status, body, ctype, extra = 200, None, "application/json", []
        if method == "GET" and path == "/api/sections":
            rows = self.db.execute("SELECT id, course, title FROM sections ORDER BY id").fetchall()
            body = {"items": [{"id": i, "course": c, "title": t, "seats_left": self.seats_left(i)} for i, c, t in rows]}
        elif method == "GET" and path == "/sections":
            rows = self.db.execute("SELECT id, course, title FROM sections ORDER BY id").fetchall()
            items = "".join(f"<li>{html.escape(c)} {html.escape(t)}: {self.seats_left(i)} seats left</li>" for i, c, t in rows)
            body, ctype = f"<main id=\"main\"><h1>Sections</h1><ul>{items}</ul></main>", "text/html; charset=utf-8"
        elif m := re.fullmatch(r"/api/sections/([\w-]+)/registrations", path):
            if method != "POST":
                status, body, extra = 405, {"title": "Method Not Allowed"}, [("Allow", "POST")]
            elif "uid" not in sess:
                status, body = 401, {"title": "Unauthorized", "detail": "sign in first"}
            elif environ.get("HTTP_X_CSRF_TOKEN") != sess.get("csrf"):
                status, body = 403, {"title": "Forbidden", "detail": "bad CSRF token"}
            else:
                status, body = self.register(m.group(1), sess["uid"])
        else:
            status, body = 404, {"title": "Not Found"}
        if status >= 400: ctype = "application/problem+json"; body = {"status": status, **body}
        data = (body if isinstance(body, str) else json.dumps(body)).encode()
        reason = {200: "OK", 201: "Created", 401: "Unauthorized", 403: "Forbidden", 404: "Not Found",
                  405: "Method Not Allowed", 409: "Conflict"}[status]
        start_response(f"{status} {reason}", [("Content-Type", ctype), ("Content-Length", str(len(data))),
                       ("X-Content-Type-Options", "nosniff"), *extra])
        return [data]


# --- tests ------------------------------------------------------------------
import unittest, threading

def call(app, method, path, cookie="", csrf=None, body=b""):
    env = {"REQUEST_METHOD": method, "PATH_INFO": path, "HTTP_COOKIE": cookie,
           "wsgi.input": io.BytesIO(body), "CONTENT_LENGTH": str(len(body))}
    if csrf: env["HTTP_X_CSRF_TOKEN"] = csrf
    out = {}
    data = b"".join(app(env, lambda s, h: out.update(status=int(s.split()[0]), headers=dict(h))))
    return out["status"], out["headers"], data.decode()

class SeatwiseTest(unittest.TestCase):
    def setUp(self):
        self.app = CourseBook()
        self.app.db.executescript("""
            INSERT INTO sections VALUES ('CMPUT404-A1', 'CMPUT 404', 'Web Applications', 2),
                                        ('CMPUT404-B1', 'CMPUT 404', 'Web <Apps> & More', 30);
            INSERT INTO users (email) VALUES ('a@example.ca'), ('b@example.ca'), ('c@example.ca');""")
    def login(self, uid):
        s = {"uid": uid, "csrf": f"tok{uid}"}
        return f"__Host-sid={self.app.sign(s)}", s["csrf"]

    def test_list_sections_json(self):
        status, h, body = call(self.app, "GET", "/api/sections")
        self.assertEqual(status, 200)
        self.assertEqual([s["seats_left"] for s in json.loads(body)["items"]], [2, 30])
    def test_html_is_escaped(self):
        _, h, body = call(self.app, "GET", "/sections")
        self.assertIn("Web &lt;Apps&gt; &amp; More", body)
        self.assertNotIn("<Apps>", body)
    def test_register_requires_login(self):
        self.assertEqual(call(self.app, "POST", "/api/sections/CMPUT404-A1/registrations")[0], 401)
    def test_register_requires_csrf(self):
        cookie, _ = self.login(1)
        self.assertEqual(call(self.app, "POST", "/api/sections/CMPUT404-A1/registrations", cookie, "wrong")[0], 403)
    def test_tampered_cookie_is_anonymous(self):
        cookie, tok = self.login(1)
        bad = cookie[:-1] + ("0" if cookie[-1] != "0" else "1")
        self.assertEqual(call(self.app, "POST", "/api/sections/CMPUT404-A1/registrations", bad, tok)[0], 401)
    def test_register_then_duplicate(self):
        cookie, tok = self.login(1)
        s, h, b = call(self.app, "POST", "/api/sections/CMPUT404-A1/registrations", cookie, tok)
        self.assertEqual((s, json.loads(b)["seats_left"]), (201, 1))
        s, h, b = call(self.app, "POST", "/api/sections/CMPUT404-A1/registrations", cookie, tok)
        self.assertEqual((s, h["Content-Type"]), (409, "application/problem+json"))
    def test_capacity_is_enforced(self):
        codes = [call(self.app, "POST", "/api/sections/CMPUT404-A1/registrations", *self.login(u))[0] for u in (1, 2, 3)]
        self.assertEqual(codes, [201, 201, 409])
    def test_unknown_section_and_method(self):
        cookie, tok = self.login(1)
        self.assertEqual(call(self.app, "POST", "/api/sections/NOPE/registrations", cookie, tok)[0], 404)
        s, h, _ = call(self.app, "GET", "/api/sections/CMPUT404-A1/registrations")
        self.assertEqual((s, h.get("Allow")), (405, "POST"))
    def test_concurrent_registrations_never_overbook(self):
        import os, tempfile
        path = os.path.join(tempfile.mkdtemp(), "seatwise.db")
        app = CourseBook(path, self.app.secret)
        app.db.executescript("INSERT INTO sections VALUES ('LAB-1', 'CMPUT 404', 'Lab', 5);" +
                             "".join(f"INSERT INTO users (email) VALUES ('u{i}@example.ca');" for i in range(20)))
        results, lock = [], threading.Lock()
        def worker(uid):
            mine = CourseBook(path, app.secret)            # one connection per thread, like separate servers
            mine.db.execute("PRAGMA busy_timeout = 5000")
            s = {"uid": uid, "csrf": "t"}
            code = call(mine, "POST", "/api/sections/LAB-1/registrations", f"__Host-sid={mine.sign(s)}", "t")[0]
            with lock: results.append(code)
        threads = [threading.Thread(target=worker, args=(u,)) for u in range(1, 21)]
        for t in threads: t.start()
        for t in threads: t.join()
        self.assertEqual((results.count(201), results.count(409)), (5, 15))
        self.assertEqual(app.seats_left("LAB-1"), 0)

if __name__ == "__main__":
    buf = io.StringIO()
    result = unittest.main(argv=["seatwise"], exit=False,
                           testRunner=unittest.TextTestRunner(stream=buf, verbosity=2)).result
    for line in buf.getvalue().splitlines():
        if " ... " in line: print(line.split(" (")[0], "...", line.rsplit(" ", 1)[1])
    print(f"{result.testsRun} tests, {len(result.failures)} failures, {len(result.errors)} errors")

test_capacity_is_enforced ... ok
test_concurrent_registrations_never_overbook ... ok
test_html_is_escaped ... ok
test_list_sections_json ... ok
test_register_requires_csrf ... ok
test_register_requires_login ... ok
test_register_then_duplicate ... ok
test_tampered_cookie_is_anonymous ... ok
test_unknown_section_and_method ... ok
9 tests, 0 failures, 0 errors

<!-- The registration widget: progressive enhancement over a server-rendered list (listing). -->
<li>
  CMPUT 404 Web Applications: <span id="left-A1" aria-live="polite">2 seats left</span>
  <form method="post" action="/sections/CMPUT404-A1/register" data-section="CMPUT404-A1">
    <input type="hidden" name="csrf" value="(token from session)">
    <button type="submit">Register for section A1</button>
  </form>
</li>
<script type="module" nonce="(per-response nonce)">
  for (const form of document.querySelectorAll('form[data-section]')) {
    form.addEventListener('submit', async e => {
      e.preventDefault();                                  // without JavaScript, the form still posts
      const id = form.dataset.section, button = form.querySelector('button');
      button.disabled = true;
      const r = await fetch(`/api/sections/${id}/registrations`, {
        method: 'POST', headers: { 'X-CSRF-Token': form.csrf.value } });
      const body = await r.json();
      document.getElementById(`left-${id.split('-')[1]}`).textContent =
        r.ok ? `${body.seats_left} seats left. You are registered.` : body.detail;
      button.disabled = r.ok;
    });
  }
</script>

# compose.yaml: the deployment for a pilot term (listing). Secrets come from the host, not the file.
services:
  web:
    image: registry.example.ca/seatwise:1.4.2
    environment:
      DATABASE_URL: postgres://seatwise@db:5432/seatwise
      SESSION_SECRET_FILE: /run/secrets/session_secret
    secrets: [session_secret]
    deploy: { replicas: 3 }
    healthcheck: { test: ["CMD", "python", "-m", "seatwise.healthcheck"], interval: 10s }
  worker:
    image: registry.example.ca/seatwise:1.4.2
    command: ["python", "-m", "seatwise.worker"]          # sends confirmation emails from the queue
  proxy:
    image: nginx:1.27
    ports: ["443:443"]
    volumes: ["./nginx.conf:/etc/nginx/nginx.conf:ro"]
  db:
    image: postgres:17
    volumes: ["pgdata:/var/lib/postgresql/data"]
  redis:
    image: redis:7
secrets:
  session_secret: { file: /etc/seatwise/session_secret }
volumes:
  pgdata: {}

# Will it survive registration day? A capacity estimate from expected traffic, using Little's law.
students, window_min = 9000, 15                 # most students try to register for the first 15 minutes
requests_per_student = 12                       # page loads, API calls, retries
peak_factor = 3                                 # the first minute is busier than average
rate = students * requests_per_student / (window_min * 60) * peak_factor
service_s = 0.060                               # mean time per request in a worker
workers_needed = rate * service_s / 0.7         # Little's law, at 70% target utilization
print(f"peak rate      {rate:6.0f} requests/s")
print(f"busy workers   {rate * service_s:6.1f} on average")
print(f"provision      {workers_needed:6.1f} -> {int(-(-workers_needed // 8))} servers of 8 workers, +1 spare")
db_writes = students * 2 / (window_min * 60) * peak_factor
print(f"database writes {db_writes:5.0f}/s at peak (registrations + email jobs): fine for one primary")

peak rate         360 requests/s
busy workers     21.6 on average
provision        30.9 -> 4 servers of 8 workers, +1 spare
database writes    60/s at peak (registrations + email jobs): fine for one primary