279 lines
10 KiB
Python
279 lines
10 KiB
Python
"""
|
||
IR -> a Jupyter notebook. Generated, never hand-authored.
|
||
|
||
## The frame, because it is the whole argument
|
||
|
||
A notebook is normally a **source file** that someone confects by hand — prose,
|
||
code and stored output braided together, diffing badly, drifting from whatever
|
||
it documents the moment either moves, with no way to tell by looking. jupytext
|
||
addresses the diffing and leaves the rest: it makes the notebook editable as
|
||
text, so you still hand-author it.
|
||
|
||
Here a notebook is a **build artifact**. The source is the OpenAPI document —
|
||
the same file the server is built from — and the notebook is regenerated from
|
||
it. Nobody edits the `.ipynb`, the same way nobody edits a `.o` file. "Is this
|
||
document current" stops being a question about somebody's diligence and becomes
|
||
a question about whether the build ran.
|
||
|
||
That is also why it suits a mixed audience rather than a data-science one. The
|
||
endpoints, their methods, their payload shapes and their status codes are facts
|
||
taken from the spec, so a PM reading it is reading the API, not somebody's
|
||
recollection of it.
|
||
|
||
## Reproducible in the strict sense
|
||
|
||
Cell ids come from position, `execution_count` is null, `outputs` is empty, and
|
||
the JSON is key-sorted. **The same IR produces byte-identical bytes.** A
|
||
notebook that changes on every build cannot be reviewed, and one that cannot be
|
||
reviewed will not be trusted.
|
||
|
||
Written against the nbformat 4 schema directly. `nbformat` is not installed on
|
||
the machines this runs on, and the schema has six required keys.
|
||
"""
|
||
|
||
import json
|
||
from pathlib import Path
|
||
|
||
METADATA = {
|
||
"kernelspec": {"display_name": "Python 3", "language": "python", "name": "python3"},
|
||
"language_info": {"name": "python"},
|
||
}
|
||
|
||
|
||
def _cell(index: int, kind: str, text: str) -> dict:
|
||
lines = text.split("\n")
|
||
source = [ln + "\n" for ln in lines[:-1]] + ([lines[-1]] if lines[-1] else [])
|
||
cell = {
|
||
"cell_type": "markdown" if kind == "md" else "code",
|
||
"id": f"cell-{index:03d}",
|
||
"metadata": {},
|
||
"source": source,
|
||
}
|
||
if kind == "code":
|
||
cell["execution_count"] = None
|
||
cell["outputs"] = []
|
||
return cell
|
||
|
||
|
||
def _example(fields: list[dict]) -> str:
|
||
"""A request body shaped like the schema, with placeholder values."""
|
||
sample = {}
|
||
for f in fields:
|
||
if f.get("pk"):
|
||
continue # the server assigns it
|
||
t = str(f.get("type", "str")).lower()
|
||
name = f["name"]
|
||
if "int" in t:
|
||
sample[name] = 0
|
||
elif "float" in t or "decimal" in t:
|
||
sample[name] = 0.0
|
||
elif "bool" in t:
|
||
sample[name] = False
|
||
elif "date" in t or "time" in t:
|
||
sample[name] = "2026-01-01T00:00:00Z"
|
||
elif "list" in t:
|
||
sample[name] = []
|
||
else:
|
||
sample[name] = f"<{name}>"
|
||
# A Python literal, not JSON. `json.dumps` writes `false`/`true`/`null`,
|
||
# which are valid *identifiers* in Python — so the cell compiles and then
|
||
# raises NameError the moment anyone runs it. Compiling is not enough of a
|
||
# check; the notebook selftest executes these cells for exactly this reason.
|
||
body = ",\n".join(f" {k!r}: {v!r}" for k, v in sorted(sample.items()))
|
||
return "{\n" + body + ",\n}" if body else "{}"
|
||
|
||
|
||
CLIENT = '''def call(method, path, params=None, body=None):
|
||
"""One request. Returns (status, parsed_body).
|
||
|
||
A non-2xx is returned rather than raised: the error body usually names the
|
||
field that was wrong, and an exception throws that away.
|
||
"""
|
||
url = BASE_URL.rstrip("/") + "/" + path.lstrip("/")
|
||
if params:
|
||
url += "?" + urllib.parse.urlencode(params)
|
||
data = json.dumps(body).encode() if body is not None else None
|
||
headers = {"Accept": "application/json", **AUTH}
|
||
if data:
|
||
headers["Content-Type"] = "application/json"
|
||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||
print(f"-> {method} {url}")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
|
||
status, raw = r.status, r.read()
|
||
except urllib.error.HTTPError as e:
|
||
status, raw = e.code, e.read()
|
||
except urllib.error.URLError as e:
|
||
print(f"<- unreachable: {e.reason}")
|
||
return None, None
|
||
try:
|
||
parsed = json.loads(raw) if raw else None
|
||
except json.JSONDecodeError:
|
||
parsed = raw.decode("utf-8", "replace")
|
||
print(f"<- {status}")
|
||
return status, parsed
|
||
|
||
|
||
def show(result, limit=1500):
|
||
status, body = result
|
||
if status is None:
|
||
return
|
||
text = body if isinstance(body, str) else json.dumps(body, indent=2)
|
||
print(text[:limit] + (f"\\n… {len(text) - limit} more" if len(text) > limit else ""))'''
|
||
|
||
|
||
def _params_cell(step: dict) -> str:
|
||
env = step.get("env_var", "API_TOKEN")
|
||
return f'''import json
|
||
import os
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
BASE_URL = os.environ.get("API_BASE_URL", "{step.get("base_url", "")}")
|
||
TOKEN = os.environ.get("{env}", "")
|
||
TIMEOUT = 30
|
||
|
||
# Read from the environment, never written here: a token pasted into a cell
|
||
# travels with every copy of this notebook from then on.
|
||
AUTH = {{"Authorization": f"Bearer {{TOKEN}}"}} if TOKEN else {{}}
|
||
|
||
print("base ", BASE_URL)
|
||
print("token", f"set ({{len(TOKEN)}} chars)" if TOKEN else "NOT SET — export {env}=…")'''
|
||
|
||
|
||
def _call_cell(step: dict) -> str:
|
||
"""The generated call. An overlay's `code` replaces this wholesale."""
|
||
method, path = step.get("method", "GET"), step.get("path", "/")
|
||
params = step.get("path_params") or []
|
||
if step.get("graphql"):
|
||
# One endpoint carries every operation, so the operation name is the
|
||
# thing worth showing, not the path.
|
||
variables = {f["name"]: f"<{f['name']}>" for f in (step.get("body_fields") or [])}
|
||
return (
|
||
f'QUERY = """{step.get("title", "query")} {{ ... }}""" '
|
||
"# fill in the selection set\n"
|
||
+ (f"VARIABLES = {variables!r}\n" if variables else "")
|
||
+ f'show(call("POST", "{path}", body={{"query": QUERY'
|
||
+ (", \"variables\": VARIABLES" if variables else "")
|
||
+ '}))'
|
||
)
|
||
lines = [f'{p.upper()} = "<{p}>" # path parameter' for p in params]
|
||
call_path = path
|
||
for p in params:
|
||
call_path = call_path.replace("{" + p + "}", f'" + str({p.upper()}) + "')
|
||
# Trim the empty concatenations a placeholder at either end leaves behind.
|
||
expr = f'"{call_path}"' if params else f'"{path}"'
|
||
expr = expr.replace(' + ""', "").replace('"" + ', "")
|
||
|
||
# Parameters that were *always* sent are not optional in practice, whatever
|
||
# the spec calls them.
|
||
always = step.get("params_always") or []
|
||
if always:
|
||
lines.append("PARAMS = " + repr({p: f"<{p}>" for p in always}))
|
||
arg = ", params=PARAMS" if always else ""
|
||
|
||
if step.get("body_fields"):
|
||
lines.append("BODY = " + _example(step["body_fields"]))
|
||
lines.append("")
|
||
lines.append(f'show(call("{method}", {expr}{arg}, body=BODY))')
|
||
else:
|
||
if lines:
|
||
lines.append("")
|
||
lines.append(f'show(call("{method}", {expr}{arg}))')
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _call_md(step: dict) -> str:
|
||
out = [f'## {step.get("method", "GET")} `{step.get("path", "/")}`']
|
||
if step.get("summary"):
|
||
out += ["", step["summary"]]
|
||
facts = []
|
||
if step.get("response_model"):
|
||
facts.append(f'returns **{step["response_model"]}**'
|
||
+ (" (a list)" if step.get("returns_list") else ""))
|
||
if step.get("request_model"):
|
||
facts.append(f'accepts **{step["request_model"]}**')
|
||
if step.get("status"):
|
||
facts.append(f'expects `{step["status"]}`')
|
||
if step.get("statuses"):
|
||
# What really came back, which is usually more than the spec promises.
|
||
facts.append("seen: " + ", ".join(f'`{c}`' for c in step["statuses"]))
|
||
if step.get("calls"):
|
||
facts.append(f'called {step["calls"]}×')
|
||
if step.get("params_sometimes"):
|
||
facts.append("sometimes sends " + ", ".join(f'`{p}`' for p in step["params_sometimes"]))
|
||
if step.get("id_formats"):
|
||
facts.append("id as " + "/".join(step["id_formats"]))
|
||
if facts:
|
||
out += ["", " · ".join(facts)]
|
||
if step.get("note"):
|
||
out += ["", step["note"]]
|
||
return "\n".join(out)
|
||
|
||
|
||
def build(spec: dict) -> dict:
|
||
"""A merged notebook spec -> the notebook, as a dict."""
|
||
blocks: list[tuple[str, str]] = []
|
||
|
||
for step in spec["steps"]:
|
||
kind = step.get("kind", "md")
|
||
before = step.get("before")
|
||
if before:
|
||
blocks.append(("md", before))
|
||
|
||
if kind == "md":
|
||
text = []
|
||
if step.get("title") and step["id"] != "intro":
|
||
text.append(f'## {step["title"]}')
|
||
elif step.get("title"):
|
||
text.append(f'# {step["title"]}')
|
||
if step.get("text"):
|
||
text += ["", step["text"]]
|
||
if step.get("table"):
|
||
text += ["", "| | fields |", "|---|---|"]
|
||
for row in step["table"]:
|
||
names = ", ".join(f'`{f}`' for f in row.get("fields", []))
|
||
text.append(f'| **{row["name"]}** | {names or "—"} |')
|
||
blocks.append(("md", "\n".join(text)))
|
||
|
||
elif kind == "params":
|
||
if step.get("title"):
|
||
blocks.append(("md", f'## {step["title"]}'))
|
||
blocks.append(("code", step.get("code") or _params_cell(step)))
|
||
|
||
elif kind == "code":
|
||
if step.get("title"):
|
||
blocks.append(("md", f'## {step["title"]}'))
|
||
code = step.get("code")
|
||
if code is None and step.get("builtin") == "client":
|
||
code = CLIENT
|
||
blocks.append(("code", code or ""))
|
||
|
||
elif kind == "call":
|
||
blocks.append(("md", _call_md(step)))
|
||
blocks.append(("code", step.get("code") or _call_cell(step)))
|
||
|
||
after = step.get("after_text")
|
||
if after:
|
||
blocks.append(("md", after))
|
||
|
||
return {
|
||
"cells": [_cell(i, k, t) for i, (k, t) in enumerate(blocks)],
|
||
"metadata": METADATA,
|
||
"nbformat": 4,
|
||
"nbformat_minor": 5,
|
||
}
|
||
|
||
|
||
def emit(spec: dict) -> str:
|
||
"""The notebook as text. Key-sorted, so the same spec gives the same bytes."""
|
||
return json.dumps(build(spec), indent=1, sort_keys=True, ensure_ascii=False) + "\n"
|
||
|
||
|
||
def write(spec: dict, path) -> Path:
|
||
path = Path(path)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(emit(spec))
|
||
return path
|