Set up for Unit 5B: integration tools

"""A local stand-in for Calder's S/4HANA order inbox, for Unit 5B.

Calder Pumpworks is a fictional company. Codes, numbers and amounts are
made up for the course. This is not SAP software and not an SAP API: it is a
small web server on your own computer that behaves like a receiving system,
including the ways a receiving system fails.

Run it from your course folder:  python unit5b/mock_sap.py
Stop it with Ctrl+C. It listens only on your own computer (127.0.0.1).

What it offers:
  GET  /health                           is the server up?
  GET  /calder/orders                    orders received so far
  GET  /calder/orders/<sales order>      one received order, OData V2 style
  POST /calder/inbound/orders            send an order message (JSON)

Faults are planned, not random, so every run behaves the same:
  HFS-PO-7781  accepted at once
  BWU-PO-5602  503 Service Unavailable twice, then accepted
  HFS-PO-7790  429 Too Many Requests once (Retry-After: 1), then accepted
  BWU-PO-5610  saved, but the first answer comes too late (client times out)
  any unknown customer                   400 Bad Request, every time
"""
import argparse
import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

API_KEY = "calder-local-key"  # a local course value, not a secret
CUSTOMERS = {"10000001": "Brennock Water Utility", "10000002": "Hallam Fluid Systems"}
PRODUCTS = {"CP-ES200", "CP-ES300", "CP-SK200"}
FAULT_PLAN = {  # customer PO number -> what goes wrong on attempts 1, 2, ...
    "BWU-PO-5602": ["503", "503"],
    "HFS-PO-7790": ["429"],
    "BWU-PO-5610": ["slow"],
}
SLOW_SECONDS = 3.0
FIRST_SALES_ORDER = 1000201


class Inbox:
    """What the receiving system remembers. One lock keeps threads apart."""

    def __init__(self):
        self.lock = threading.Lock()
        self.attempts = {}  # customer PO -> attempts seen
        self.by_key = {}    # idempotency key -> (body, status, answer)
        self.orders = {}    # sales order -> order
        self.next_number = FIRST_SALES_ORDER

    def receive(self, key, body):
        """Return (status, answer, delay, extra headers) for one POST."""
        with self.lock:
            po = str(body.get("PurchaseOrderByCustomer", ""))
            attempt = self.attempts[po] = self.attempts.get(po, 0) + 1
            if key in self.by_key:
                old_body, status, answer = self.by_key[key]
                if old_body != body:
                    return 422, {"error": "Idempotency-Key reused with a different message"}, 0, {}
                return status, answer, 0, {"Idempotent-Replay": "true"}
            plan = FAULT_PLAN.get(po, [])
            fault = plan[attempt - 1] if attempt <= len(plan) else ""
            if fault == "503":
                return 503, {"error": "Service unavailable, try again later"}, 0, {}
            if fault == "429":
                return 429, {"error": "Too many requests"}, 0, {"Retry-After": "1"}
            problem = self.validate(body)
            if problem:
                return 400, {"error": problem}, 0, {}
            number = str(self.next_number)
            self.next_number += 1
            order = {"SalesOrder": number, "SalesOrganization": "CPUS", "SoldToParty": body["SoldToParty"],
                     "SoldToPartyName": CUSTOMERS[body["SoldToParty"]],
                     "PurchaseOrderByCustomer": po, "to_Item": body["Items"]}
            self.orders[number] = order
            answer = {"SalesOrder": number, "PurchaseOrderByCustomer": po}
            self.by_key[key] = (body, 201, answer)
            return 201, answer, SLOW_SECONDS if fault == "slow" else 0, {}

    @staticmethod
    def validate(body):
        if body.get("SoldToParty") not in CUSTOMERS:
            return f"Customer {body.get('SoldToParty')} does not exist"
        items = body.get("Items") or []
        if not items:
            return "Order has no items"
        for item in items:
            if item.get("Material") not in PRODUCTS:
                return f"Product {item.get('Material')} does not exist"
            if not isinstance(item.get("RequestedQuantity"), int) or item["RequestedQuantity"] <= 0:
                return f"Quantity for {item.get('Material')} must be a positive whole number"
        return ""


def make_handler(inbox, quiet=False):
    class Handler(BaseHTTPRequestHandler):
        def log_message(self, fmt, *args):  # one short line per request
            if not quiet:
                print(f"  mock  {self.command} {self.path} -> {args[1] if len(args) > 1 else ''}")

        def send_json(self, status, data, headers=None):
            raw = json.dumps(data, indent=2).encode("utf-8")
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(raw)))
            for name, value in (headers or {}).items():
                self.send_header(name, value)
            self.end_headers()
            try:
                self.wfile.write(raw)
            except OSError:
                pass  # the client stopped waiting; the work is already saved

        def authorized(self):
            if self.headers.get("APIKey") == API_KEY:
                return True
            self.send_json(401, {"error": "Missing or wrong APIKey header"})
            return False

        def do_GET(self):
            if self.path == "/health":
                return self.send_json(200, {"status": "up"})
            if not self.authorized():
                return
            if self.path == "/calder/orders":
                with inbox.lock:
                    rows = [{"SalesOrder": o["SalesOrder"], "PurchaseOrderByCustomer": o["PurchaseOrderByCustomer"],
                             "SoldToParty": o["SoldToParty"]} for o in inbox.orders.values()]
                return self.send_json(200, {"d": {"results": rows}})
            prefix = "/calder/orders/"
            if self.path.startswith(prefix):
                with inbox.lock:
                    order = inbox.orders.get(self.path[len(prefix):])
                if order:
                    return self.send_json(200, {"d": order})
                return self.send_json(404, {"error": "No such sales order"})
            self.send_json(404, {"error": "Unknown path"})

        def do_POST(self):
            if self.path != "/calder/inbound/orders":
                return self.send_json(404, {"error": "Unknown path"})
            if not self.authorized():
                return
            key = self.headers.get("Idempotency-Key", "")
            if not key:
                return self.send_json(400, {"error": "Idempotency-Key header is required"})
            try:
                length = int(self.headers.get("Content-Length", "0"))
                body = json.loads(self.rfile.read(length) or b"{}")
            except (ValueError, json.JSONDecodeError):
                return self.send_json(400, {"error": "Body is not valid JSON"})
            status, answer, delay, headers = inbox.receive(key, body)
            if delay:
                time.sleep(delay)
            self.send_json(status, answer, headers)

    return Handler


def start(port=0, quiet=True):
    """Start the mock in a background thread; return (server, inbox). Used by the check script."""
    inbox = Inbox()
    server = ThreadingHTTPServer(("127.0.0.1", port), make_handler(inbox, quiet))
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server, inbox


def main():
    parser = argparse.ArgumentParser(description="Local mock of Calder's order inbox (fictional).")
    parser.add_argument("--port", type=int, default=8765, help="port to listen on (default 8765)")
    args = parser.parse_args()
    server = ThreadingHTTPServer(("127.0.0.1", args.port), make_handler(Inbox()))
    print(f"Calder mock order inbox on http://127.0.0.1:{args.port}  (fictional data, Ctrl+C to stop)")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nStopped.")


if __name__ == "__main__":
    main()

Calder mock order inbox on http://127.0.0.1:8765  (fictional data, Ctrl+C to stop)

"""A tiny durable queue with retries and a dead-letter queue, for Unit 5B.

Calder Pumpworks is fictional; the messages are made-up distributor and
customer orders. The queue is one SQLite file (unit5b/queue.db), so messages
survive if you stop the worker halfway.

Run these from your course folder, with the mock running in another terminal:
  python unit5b/queue_worker.py reset            empty the queue
  python unit5b/queue_worker.py enqueue --sample put five sample orders in
  python unit5b/queue_worker.py run              send them, retrying as needed
  python unit5b/queue_worker.py status           show every message's state
  python unit5b/queue_worker.py replay <id>      move a dead letter back to the queue
"""
import argparse
import json
import random
import sqlite3
import sys
import time
import urllib.error
import urllib.request
import uuid
from pathlib import Path

DB = Path(__file__).resolve().parent / "queue.db"
BASE_URL = "http://127.0.0.1:8765"
API_KEY = "calder-local-key"  # the mock's local course value, not a secret
MAX_ATTEMPTS = 4              # first try plus three retries
BASE_DELAY = 0.5              # seconds before the first retry
MAX_DELAY = 4.0               # never wait longer than this between tries
TIMEOUT = 1.5                 # seconds to wait for an answer
RETRYABLE = {429, 500, 502, 503, 504}
SEED = 42                     # fixed, so the jitter is the same on every run

SAMPLE = [  # fictional inbound orders, as a distributor or customer system would send them
    {"PurchaseOrderByCustomer": "HFS-PO-7781", "SoldToParty": "10000002",
     "Items": [{"Material": "CP-SK200", "RequestedQuantity": 60}]},
    {"PurchaseOrderByCustomer": "BWU-PO-5602", "SoldToParty": "10000001",
     "Items": [{"Material": "CP-ES200", "RequestedQuantity": 2}]},
    {"PurchaseOrderByCustomer": "HFS-PO-7790", "SoldToParty": "10000002",
     "Items": [{"Material": "CP-ES200", "RequestedQuantity": 4}]},
    {"PurchaseOrderByCustomer": "BWU-PO-5610", "SoldToParty": "10000001",
     "Items": [{"Material": "CP-SK200", "RequestedQuantity": 20}]},
    {"PurchaseOrderByCustomer": "XYZ-PO-0001", "SoldToParty": "10009999",
     "Items": [{"Material": "CP-ES300", "RequestedQuantity": 1}]},
]


def connect(db=DB):
    con = sqlite3.connect(db)
    con.execute("""CREATE TABLE IF NOT EXISTS messages (
        id INTEGER PRIMARY KEY, idem_key TEXT UNIQUE, body TEXT, state TEXT,
        attempts INTEGER DEFAULT 0, last_result TEXT DEFAULT '', sales_order TEXT DEFAULT '')""")
    return con


def enqueue(con, body):
    """Add one message. The idempotency key is made once, here, and reused on every retry."""
    con.execute("INSERT INTO messages (idem_key, body, state) VALUES (?, ?, 'queued')",
                (str(uuid.uuid4()), json.dumps(body)))
    con.commit()


def send(base_url, key, body):
    """POST one message. Return (status or None, answer dict, Retry-After seconds or None)."""
    request = urllib.request.Request(
        f"{base_url}/calder/inbound/orders", data=json.dumps(body).encode("utf-8"), method="POST",
        headers={"Content-Type": "application/json", "APIKey": API_KEY, "Idempotency-Key": key})
    try:
        with urllib.request.urlopen(request, timeout=TIMEOUT) as resp:
            answer = json.loads(resp.read() or b"{}")
            answer["replayed"] = resp.headers.get("Idempotent-Replay") == "true"
            return resp.status, answer, None
    except urllib.error.HTTPError as err:
        retry_after = err.headers.get("Retry-After")
        return err.code, json.loads(err.read() or b"{}"), float(retry_after) if retry_after else None
    except (urllib.error.URLError, TimeoutError, OSError) as err:
        reason = getattr(err, "reason", err)
        return None, {"error": f"no answer ({type(reason).__name__})"}, None


def backoff(attempt, rng):
    """Exponential backoff with jitter: about 0.5 s, 1 s, 2 s ... never above MAX_DELAY."""
    delay = min(MAX_DELAY, BASE_DELAY * 2 ** (attempt - 1))
    return round(delay * (0.5 + rng.random() / 2), 2)


def run(con, base_url=BASE_URL, out=print, sleep=time.sleep):
    """Work through the queue once. Return counts of each final state."""
    rng = random.Random(SEED)
    rows = con.execute("SELECT id, idem_key, body, attempts FROM messages WHERE state = 'queued' ORDER BY id").fetchall()
    for msg_id, key, raw, attempts in rows:
        body = json.loads(raw)
        po = body["PurchaseOrderByCustomer"]
        while True:
            attempts += 1
            status, answer, retry_after = send(base_url, key, body)
            shown = status if status else "timeout"
            if status in (200, 201):
                note = " (replayed: already saved on an earlier try)" if answer["replayed"] else ""
                out(f"  #{msg_id} {po}  try {attempts}: {shown} -> sales order {answer['SalesOrder']}{note}")
                con.execute("UPDATE messages SET state='done', attempts=?, last_result=?, sales_order=? WHERE id=?",
                            (attempts, str(status), answer["SalesOrder"], msg_id))
                break
            retryable = status is None or status in RETRYABLE
            if not retryable or attempts >= MAX_ATTEMPTS:
                why = "not retryable" if not retryable else f"gave up after {attempts} tries"
                out(f"  #{msg_id} {po}  try {attempts}: {shown} {answer.get('error', '')} -> dead letter ({why})")
                con.execute("UPDATE messages SET state='dead', attempts=?, last_result=? WHERE id=?",
                            (attempts, f"{shown} {answer.get('error', '')}".strip(), msg_id))
                break
            wait = retry_after if retry_after is not None else backoff(attempts, rng)
            source = "Retry-After" if retry_after is not None else "backoff"
            out(f"  #{msg_id} {po}  try {attempts}: {shown} -> retry in {wait:.2f} s ({source})")
            con.execute("UPDATE messages SET attempts=?, last_result=? WHERE id=?", (attempts, str(shown), msg_id))
            con.commit()
            sleep(wait)
        con.commit()
    return dict(con.execute("SELECT state, COUNT(*) FROM messages GROUP BY state").fetchall())


def status(con):
    print(f"  {'id':>3}  {'customer PO':<12} {'state':<7} {'tries':>5}  {'sales order':<11} last result")
    for row in con.execute("SELECT id, body, state, attempts, sales_order, last_result FROM messages ORDER BY id"):
        po = json.loads(row[1])["PurchaseOrderByCustomer"]
        print(f"  {row[0]:>3}  {po:<12} {row[2]:<7} {row[3]:>5}  {row[4] or '-':<11} {row[5]}")


def main():
    parser = argparse.ArgumentParser(description="Durable queue with retries and dead letters (fictional data).")
    parser.add_argument("command", choices=["reset", "enqueue", "run", "status", "replay"])
    parser.add_argument("id", nargs="?", type=int, help="message id, for replay")
    parser.add_argument("--sample", action="store_true", help="enqueue the five sample orders")
    args = parser.parse_args()
    con = connect()
    if args.command == "reset":
        con.execute("DELETE FROM messages")
        con.commit()
        print("Queue is empty.")
    elif args.command == "enqueue":
        if not args.sample:
            sys.exit("Use: enqueue --sample")
        for body in SAMPLE:
            enqueue(con, body)
        print(f"Queued {len(SAMPLE)} sample orders (fictional).")
    elif args.command == "run":
        counts = run(con)
        print(f"\nDone: {counts.get('done', 0)} delivered, {counts.get('dead', 0)} in the dead-letter queue, "
              f"{counts.get('queued', 0)} still queued.")
    elif args.command == "status":
        status(con)
    elif args.command == "replay":
        changed = con.execute("UPDATE messages SET state='queued', attempts=0 WHERE id=? AND state='dead'",
                              (args.id,)).rowcount
        con.commit()
        print(f"Message {args.id} is queued again." if changed else f"Message {args.id} is not a dead letter.")


if __name__ == "__main__":
    main()

Queue is empty.
Queued 5 sample orders (fictional).
  #1 HFS-PO-7781  try 1: 201 -> sales order 1000201
  #2 BWU-PO-5602  try 1: 503 -> retry in 0.41 s (backoff)
  #2 BWU-PO-5602  try 2: 503 -> retry in 0.51 s (backoff)
  #2 BWU-PO-5602  try 3: 201 -> sales order 1000202
  #3 HFS-PO-7790  try 1: 429 -> retry in 1.00 s (Retry-After)
  #3 HFS-PO-7790  try 2: 201 -> sales order 1000203
  #4 BWU-PO-5610  try 1: timeout -> retry in 0.32 s (backoff)
  #4 BWU-PO-5610  try 2: 201 -> sales order 1000204 (replayed: already saved on an earlier try)
  #5 XYZ-PO-0001  try 1: 400 Customer 10009999 does not exist -> dead letter (not retryable)

Done: 4 delivered, 1 in the dead-letter queue, 0 still queued.

   id  customer PO  state   tries  sales order last result
    1  HFS-PO-7781  done        1  1000201     201
    2  BWU-PO-5602  done        3  1000202     201
    3  HFS-PO-7790  done        2  1000203     201
    4  BWU-PO-5610  done        2  1000204     201
    5  XYZ-PO-0001  dead        1  -           400 Customer 10009999 does not exist

# Calder mock order inbox (fictional). Start it first: python unit5b/mock_sap.py
@base = http://127.0.0.1:8765
@key = calder-local-key

### 1. Is the mock up?
GET {{base}}/health

### 2. Send one order. Send it twice: the second answer is a replay, not a new order.
POST {{base}}/calder/inbound/orders
APIKey: {{key}}
Content-Type: application/json
Idempotency-Key: try-it-0001

{"PurchaseOrderByCustomer": "HFS-PO-7781", "SoldToParty": "10000002",
 "Items": [{"Material": "CP-SK200", "RequestedQuantity": 60}]}

### 3. List the orders the mock has received
GET {{base}}/calder/orders
APIKey: {{key}}

### 4. A call without the APIKey header is refused (401)
GET {{base}}/calder/orders

# SAP Business Accelerator Hub sandbox (SAP's demo data, not Calder's).
# Reads SAP_API_KEY from the .env file in this same folder. Never paste the key here.

### One sales order number from the sandbox
GET https://sandbox.api.sap.com/s4hanacloud/sap/opu/odata/sap/API_SALES_ORDER_SRV/A_SalesOrder?$top=1&$select=SalesOrder&$format=json
APIKey: {{$dotenv SAP_API_KEY}}
Accept: application/json

{"PurchaseOrderByCustomer": "HFS-PO-7781", "SoldToParty": "10000002",
 "Items": [{"Material": "CP-SK200", "RequestedQuantity": 60}]}

"""Check that your computer is ready for Unit 5B (integration architecture).

Run it from your course folder:  python check_unit5b.py
It starts its own copy of the Calder mock on a free port, sends the five
sample orders through a throw-away queue, and checks the results. It does
not touch unit5b/queue.db, needs no internet and prints no keys.
"""
import json
import os
import shutil
import subprocess
import sys
import tempfile
import urllib.error
import urllib.request
from pathlib import Path

UNIT = Path("unit5b")
problems = 0


def report(ok: bool, label: str, fix: str = "", optional: bool = False) -> None:
    """Print one line: OK, MISSING (must fix) or LATER (optional for now)."""
    global problems
    if ok:
        print(f"  OK       {label}")
    elif optional:
        print(f"  LATER    {label}  ->  {fix}")
    else:
        problems += 1
        print(f"  MISSING  {label}  ->  {fix}")


def get_status(url: str, headers: dict) -> int:
    try:
        with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=5) as resp:
            return resp.status
    except urllib.error.HTTPError as err:
        return err.code


print("\n1. Python")
v = sys.version_info
report(v >= (3, 10), f"Python {v.major}.{v.minor}.{v.micro}", "install Python 3.10 or newer (course setup topic)")
report(sys.prefix != sys.base_prefix, "virtual environment is active", "activate .venv (course setup topic, Step 4)")

print("\n2. Unit 5B files")
files = ["mock_sap.py", "queue_worker.py", "calder.http"]
for name in files:
    report((UNIT / name).exists(), f"unit5b/{name}", "create it (this topic, Steps 2 to 4)")

print("\n3. Mock and queue")
if (UNIT / "mock_sap.py").exists() and (UNIT / "queue_worker.py").exists():
    sys.path.insert(0, str(UNIT.resolve()))
    import mock_sap
    import queue_worker

    server, inbox = mock_sap.start(port=0)
    base = f"http://127.0.0.1:{server.server_address[1]}"
    report(get_status(f"{base}/health", {}) == 200, "mock answers on this computer")
    report(get_status(f"{base}/calder/orders", {}) == 401, "mock refuses a call without the APIKey header")
    with tempfile.TemporaryDirectory() as tmp:
        con = queue_worker.connect(Path(tmp) / "check.db")
        for body in queue_worker.SAMPLE:
            queue_worker.enqueue(con, body)
        lines = []
        counts = queue_worker.run(con, base_url=base, out=lines.append, sleep=lambda s: None)
        con.close()
    report(counts.get("done") == 4 and counts.get("dead") == 1,
           f"queue delivered {counts.get('done', 0)}, dead-lettered {counts.get('dead', 0)} (expected 4 and 1)",
           "copy queue_worker.py and mock_sap.py again from this topic")
    retried = sum("retry in" in line for line in lines)
    report(retried == 4, f"{retried} retries for the planned faults (expected 4)", "copy both scripts again")
    report(len(inbox.orders) == 4, f"{len(inbox.orders)} sales orders created, no duplicates (expected 4)",
           "check that each message keeps one Idempotency-Key for all its tries")
    server.shutdown()
else:
    report(False, "mock and queue test", "create both scripts first (Steps 2 and 3)")

print("\n4. HTTP clients")
curl = shutil.which("curl.exe" if os.name == "nt" else "curl") or shutil.which("curl")
report(bool(curl), "curl is installed", "Windows 10 and 11 include it; on macOS and Linux check with your package manager", optional=True)
code = shutil.which("code")
extensions = ""
if code:
    try:
        extensions = subprocess.run([code, "--list-extensions"], capture_output=True, text=True,
                                    timeout=60).stdout.lower()
    except (OSError, subprocess.SubprocessError):
        extensions = ""
report("humao.rest-client" in extensions, "VS Code REST Client extension",
       "install it in VS Code (Step 1), or skip this if you use curl", optional=True)

print()
if problems:
    print(f"{problems} item(s) to fix. Fix them in order, then run this again.")
    sys.exit(1)
print("All set. You are ready for Unit 5B.")

1. Python
  OK       Python 3.13.18
  OK       virtual environment is active

2. Unit 5B files
  OK       unit5b/mock_sap.py
  OK       unit5b/queue_worker.py
  OK       unit5b/calder.http

3. Mock and queue
  OK       mock answers on this computer
  OK       mock refuses a call without the APIKey header
  OK       queue delivered 4, dead-lettered 1 (expected 4 and 1)
  OK       4 retries for the planned faults (expected 4)
  OK       4 sales orders created, no duplicates (expected 4)

4. HTTP clients
  OK       curl is installed
  OK       VS Code REST Client extension

All set. You are ready for Unit 5B.

git add unit5b/mock_sap.py unit5b/queue_worker.py unit5b/calder.http check_unit5b.py
git commit -m "Add Unit 5B mock inbox, queue and check"