#!/usr/bin/env python3 """Original REST/HTTP teaching lab. Python 3.11+; standard library only. Real HTTP on a temporary 127.0.0.1 port, fictional in-memory tickets. No model, credentials, persistent files or external requests. Not a production server. """ import argparse import copy import json from contextlib import contextmanager from http.client import HTTPConnection from http.server import BaseHTTPRequestHandler, HTTPServer from threading import Thread CREATE_ATTEMPTS = 2 REFRESH_BEFORE_SECOND_EDIT = False CASES = ("read", "create-repeat", "replace-repeat", "delete-repeat", "stale-edit", "missing-precondition", "partial-replacement") def encode(value): return json.dumps(value, sort_keys=True, ensure_ascii=False).encode("utf-8") def entry(title="Pump check", status="open"): return {"body": encode({"title": title, "status": status}), "version": 1} def tag(ticket_id, row): return f'"{ticket_id}-v{row["version"]}"' def snapshot(server): return {key: {"representation": json.loads(row["body"]), "etag": tag(key, row)} for key, row in sorted(server.tickets.items())} class TicketHandler(BaseHTTPRequestHandler): # HTTP/1.0 closes each exchange; this deliberately small lab is sequential. def log_message(self, *_): pass def reply(self, status, value=None, headers=None, raw=None): body = raw if raw is not None else (encode(value) if value is not None else b"") self.send_response(status) self.send_header("Cache-Control", "no-store") if body: self.send_header("Content-Type", "application/json") if status != 204: self.send_header("Content-Length", str(len(body))) for name, value in (headers or {}).items(): self.send_header(name, value) self.end_headers() if body: self.wfile.write(body) def read_representation(self): try: size = int(self.headers.get("Content-Length", "0")) if self.headers.get("Content-Type") != "application/json" or not 0 < size <= 4096: raise ValueError("Expected a small JSON representation") raw = self.rfile.read(size) value = json.loads(raw) if not isinstance(value, dict) or set(value) != {"title", "status"}: raise ValueError("Supply the complete title/status representation") if not isinstance(value["title"], str) or not value["title"].strip() or len(value["title"]) > 120: raise ValueError("Title must contain 1 to 120 characters") if value["status"] not in ("open", "queued", "closed"): raise ValueError("Status must be open, queued or closed") return raw except (ValueError, TypeError, UnicodeDecodeError): self.reply(400, {"error": "invalid-representation", "required": "title and status"}) return None def dispatch(self): rows = self.server.tickets ticket_id = self.path.removeprefix("/tickets/") item_path = self.path.startswith("/tickets/") and "/" not in ticket_id row = rows.get(ticket_id) if item_path else None if self.command == "POST" and self.path == "/tickets": raw = self.read_representation() if raw is None: return ticket_id = f"T-{self.server.next_id}" self.server.next_id += 1 rows[ticket_id] = {"body": raw, "version": 1} self.reply(201, headers={"Location": f"/tickets/{ticket_id}", "ETag": tag(ticket_id, rows[ticket_id])}, raw=raw) return if not row: self.reply(404, {"error": "not-found"}) return if self.command == "GET": self.reply(200, headers={"ETag": tag(ticket_id, row)}, raw=row["body"]) elif self.command == "PUT": raw = self.read_representation() if raw is None: return expected = self.headers.get("If-Match") if expected is None: self.reply(428, {"error": "precondition-required", "next": "Read and review the resource; supply its ETag in If-Match"}) return # This teaching contract accepts exactly one strong tag from GET, # not the full HTTP syntax (tag lists and wildcard are out of scope). if expected != tag(ticket_id, row): self.reply(412, {"error": "stale-representation", "next": "Read the current resource and reconcile your intended change"}) return if raw != row["body"]: row["body"] = raw # Store the submitted representation unchanged. row["version"] += 1 self.reply(200, headers={"ETag": tag(ticket_id, row)}, raw=row["body"]) elif self.command == "DELETE": del rows[ticket_id] self.reply(204) else: self.reply(405, {"error": "method-not-allowed"}, {"Allow": "GET, PUT, DELETE"}) do_GET = do_POST = do_PUT = do_DELETE = dispatch @contextmanager def local_service(): server = HTTPServer(("127.0.0.1", 0), TicketHandler) server.tickets = {"T-104": entry()} server.next_id = 105 worker = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.02}, daemon=True) worker.start() try: yield server finally: server.shutdown() worker.join(timeout=2) server.server_close() def run_case(name): with local_service() as server: initial = snapshot(server) trace = [] def request(method, path="/tickets/T-104", value=None, etag=None): headers = {"Content-Type": "application/json"} if value is not None else {} if etag is not None: headers["If-Match"] = etag connection = HTTPConnection("127.0.0.1", server.server_port, timeout=2) try: connection.request(method, path, body=encode(value) if value is not None else None, headers=headers) response = connection.getresponse() raw = response.read() result = {"request": {"method": method, "path": path, "headers": headers, "body": value}, "response": {"status": response.status, "headers": {k: v for k, v in response.getheaders() if k in ("ETag", "Location", "Allow")}, "body": json.loads(raw) if raw else None}, "state_after": snapshot(server)} trace.append(copy.deepcopy(result)) return result["response"] finally: connection.close() if name == "read": request("GET") request("GET") elif name == "create-repeat": for _ in range(CREATE_ATTEMPTS): request("POST", "/tickets", {"title": "Valve check", "status": "open"}) elif name == "replace-repeat": current = request("GET") for _ in range(2): request("PUT", value={"title": "Pump check", "status": "queued"}, etag=current["headers"]["ETag"]) elif name == "delete-repeat": request("DELETE") request("DELETE") elif name == "stale-edit": old = request("GET") request("PUT", value={"title": "Pump check: reviewed scope", "status": "queued"}, etag=old["headers"]["ETag"]) etag = request("GET")["headers"]["ETag"] if REFRESH_BEFORE_SECOND_EDIT else old["headers"]["ETag"] request("PUT", value={"title": "Pump check", "status": "closed"}, etag=etag) request("GET") elif name == "missing-precondition": request("PUT", value={"title": "Pump check", "status": "queued"}) elif name == "partial-replacement": current = request("GET") request("PUT", value={"status": "closed"}, etag=current["headers"]["ETag"]) return {"case": name, "initial": initial, "trace": trace, "final": snapshot(server)} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("case", choices=("compare",) + CASES, nargs="?", default="compare") args = parser.parse_args() names = CASES if args.case == "compare" else (args.case,) print(json.dumps({"scope": "Real loopback HTTP; fictional in-memory resources; no authentication, model or external service", "results": [run_case(name) for name in names]}, indent=2)) if __name__ == "__main__": main()