Coverage for diagrams / api_specs / generate.py: 100.00%
774 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1#!/usr/bin/env python3
2"""Render the API spec sheets: one Markdown page per GCO HTTP service.
4The source of truth is what FastAPI itself generates. ``scripts/generate_openapi.py``
5asks each application for ``app.openapi()`` and commits the result under
6``docs/openapi/<service>.json`` — the document behind the service's automatic
7Swagger ``/docs`` page — and ``tests/test_api_docs_coverage.py`` fails when
8those documents drift from the live routes. This catalogue turns each document
9into two human-facing renderings:
11* ``diagrams/api_specs/<service>.md`` — a Markdown spec sheet: the endpoint
12 table, every operation's parameters, request body and responses, and every
13 component schema as a property table. It renders on GitHub and, staged into
14 the wiki's source tree by ``scripts/build_wiki.py``, as a page of the
15 Zensical-built wiki, so the API reference is embedded in the documentation
16 without a hand-maintained copy.
17* The Swagger UI console FastAPI serves at ``/docs`` — ``--swagger-ui-dir``
18 writes FastAPI's own ``get_swagger_ui_html`` page per service, pointing at a
19 copy of the document and at a self-hosted ``swagger-ui-dist`` (the pinned npm
20 dependency), so the Pages site keeps making zero third-party requests. That
21 tree is built at deploy time by ``pages.yml`` and is not committed.
23Everything is a deterministic function of the committed documents, so
24freshness is verified by regenerating in memory and comparing (``--check``);
25unlike the code flowcharts there is no rasteriser and nothing to stamp. The
26render path is stdlib-only, which keeps ``python -S diagrams/generate.py
27--check`` working without site-packages; FastAPI is imported only to build the
28Swagger tree.
30Usage::
32 python diagrams/api_specs/generate.py # rewrite the sheets + README
33 python diagrams/api_specs/generate.py --check # exit 1 if any committed sheet is stale
34 python diagrams/api_specs/generate.py --swagger-ui-dir site/swagger \\
35 --swagger-assets node_modules/swagger-ui-dist # build the interactive consoles
36"""
38from __future__ import annotations
40import argparse
41import json
42import re
43import shutil
44import sys
45import tomllib
46from pathlib import Path
47from typing import Any
49#: The checkout this generator renders for; tests point the functions at a
50#: temporary root instead, which is why every entry point takes one.
51REPO_ROOT = Path(__file__).resolve().parents[2]
53#: Path prefix of the Swagger UI consoles on the Pages site (``/swagger/<service>/``).
54#: Distinct from the wiki's ``api/`` prefix, where the Markdown sheets live.
55SWAGGER_SITE_PREFIX = "swagger"
57#: The regeneration command every generated file names.
58REGENERATION_COMMAND = "python diagrams/generate.py --api-only"
60#: Files ``--swagger-ui-dir`` requires from the ``swagger-ui-dist`` package.
61SWAGGER_ASSET_FILES: tuple[str, ...] = (
62 "swagger-ui-bundle.js",
63 "swagger-ui.css",
64 "favicon-32x32.png",
65)
66#: Redistribution notices copied alongside the assets when the package ships them.
67SWAGGER_NOTICE_FILES: tuple[str, ...] = ("LICENSE", "NOTICE", "swagger-ui-bundle.js.LICENSE.txt")
69#: HTTP methods in the order a reader expects, not alphabetically.
70_METHOD_ORDER: tuple[str, ...] = (
71 "get",
72 "post",
73 "put",
74 "patch",
75 "delete",
76 "head",
77 "options",
78 "trace",
79)
81#: The wiki's configuration, the one place the site and repository URLs live.
82SITE_CONFIG_FILE = "zensical.toml"
83_REF_PREFIX = "#/components/schemas/"
85#: The interaction diagram rendered next to the sheets.
86TOPOLOGY_FILE = "api-topology.svg"
88#: How each kind of document came to be, keyed by ``x-gco-source.kind`` (a
89#: FastAPI export carries no source block). ``noun`` labels the sheet header,
90#: ``origin`` finishes the sentence "the ... this sheet is rendered from".
91_KINDS: dict[str, tuple[str, str]] = {
92 "fastapi": (
93 "Service",
94 "FastAPI `app.openapi()` export (`scripts/generate_openapi.py`) this sheet is rendered from",
95 ),
96 "aws-api-gateway": (
97 "API Gateway",
98 "document `scripts/generate_api_gateway_openapi.py` reads out of the synthesized "
99 "CDK stack; this sheet is rendered from it",
100 ),
101 "cluster-gateway": (
102 "Gateway",
103 "document `scripts/generate_cluster_gateway_openapi.py` composes from the HTTPRoute "
104 "manifest and the service documents; this sheet is rendered from it",
105 ),
106}
107#: Reading order of the catalogue index and the diagram: front doors first.
108_KIND_ORDER: tuple[str, ...] = ("aws-api-gateway", "cluster-gateway", "fastapi")
111class SpecSheetError(RuntimeError):
112 """A document or the site configuration cannot be rendered faithfully."""
115def _dict(value: Any) -> dict[str, Any]:
116 """A JSON object as itself, anything else as an empty mapping."""
117 return value if isinstance(value, dict) else {}
120# ─── inputs ──────────────────────────────────────────────────────────────────
123def service_names(openapi_dir: Path) -> list[str]:
124 """The services with a committed OpenAPI document, by filename stem."""
125 names = sorted(path.stem for path in openapi_dir.glob("*.json"))
126 if not names:
127 raise SpecSheetError(f"no OpenAPI documents under {openapi_dir}")
128 return names
131def load_document(service: str, openapi_dir: Path) -> dict[str, Any]:
132 document = json.loads((openapi_dir / f"{service}.json").read_text(encoding="utf-8"))
133 if not isinstance(document, dict) or not isinstance(document.get("paths"), dict):
134 raise SpecSheetError(f"{service}: OpenAPI document has no paths object")
135 return document
138def document_kind(document: dict[str, Any]) -> str:
139 """``fastapi``, ``aws-api-gateway`` or ``cluster-gateway`` — what produced the document."""
140 kind = _dict(document.get("x-gco-source")).get("kind") or "fastapi"
141 if kind not in _KINDS:
142 raise SpecSheetError(f"unknown x-gco-source.kind {kind!r}")
143 return str(kind)
146def catalogue_order(documents: dict[str, dict[str, Any]]) -> list[str]:
147 """Document names, front doors first, then alphabetical within a kind."""
148 return sorted(
149 documents, key=lambda name: (_KIND_ORDER.index(document_kind(documents[name])), name)
150 )
153def read_site_config(zensical_toml: Path) -> tuple[str, str]:
154 """``(site_url, repo_url)`` from zensical.toml — the one place both are declared.
156 Read with the stdlib ``tomllib`` so the render path needs nothing beyond the
157 standard library, and so a fork's ``scripts/migrate_fork.py`` rewrite of
158 zensical.toml is the only edit needed for the generated links to follow.
159 """
160 try:
161 with zensical_toml.open("rb") as handle:
162 project = tomllib.load(handle).get("project")
163 except tomllib.TOMLDecodeError as exc:
164 raise SpecSheetError(f"{zensical_toml}: not valid TOML ({exc})") from exc
165 site = project.get("site_url") if isinstance(project, dict) else None
166 repo = project.get("repo_url") if isinstance(project, dict) else None
167 if not isinstance(site, str) or not isinstance(repo, str) or not site or not repo:
168 raise SpecSheetError(
169 f"{zensical_toml}: [project] site_url and repo_url must both be declared"
170 )
171 return site.rstrip("/") + "/", repo.rstrip("/")
174# ─── Markdown primitives ─────────────────────────────────────────────────────
177def slugify(heading: str) -> str:
178 """GitHub-compatible heading anchor (python-markdown's toc agrees for ASCII).
180 Lower-case, code spans unwrapped, anything but letters, digits, spaces,
181 hyphens and underscores dropped, spaces to hyphens. ``GET /api/v1/jobs/{id}``
182 becomes ``get-apiv1jobsid``.
183 """
184 text = heading.replace("`", "").lower()
185 text = re.sub(r"[^a-z0-9 _-]", "", text)
186 return re.sub(r"\s+", "-", text.strip())
189def cell(text: object) -> str:
190 """One table cell: whitespace collapsed, pipes escaped, empty shown as a dash."""
191 collapsed = " ".join(str(text).split()) if text is not None else ""
192 return collapsed.replace("|", "\\|") or "—"
195def code(text: str) -> str:
196 return f"`{text}`"
199def table(headers: list[str], rows: list[list[str]]) -> list[str]:
200 """A Markdown table; every row is padded to the header width (MD056)."""
201 width = len(headers)
202 lines = ["| " + " | ".join(headers) + " |", "|" + "---|" * width]
203 for row in rows:
204 padded = [*row, *(["—"] * (width - len(row)))][:width]
205 lines.append("| " + " | ".join(padded) + " |")
206 return lines
209def _aligned_commands(*entries: tuple[str, str]) -> list[str]:
210 """Shell lines whose trailing ``# comment`` starts in one column.
212 The column is the longest command plus two spaces, so a renamed or added
213 command re-aligns the whole block instead of overflowing a fixed width.
214 """
215 width = max(len(command) for command, _comment in entries)
216 return [f"{command:<{width}} # {comment}" for command, comment in entries]
219# ─── schema rendering ────────────────────────────────────────────────────────
222def _ref_name(ref: str) -> str:
223 if not ref.startswith(_REF_PREFIX):
224 raise SpecSheetError(f"unsupported $ref outside components/schemas: {ref}")
225 return ref[len(_REF_PREFIX) :]
228def schema_link(name: str) -> str:
229 return f"[{code(name)}](#{schema_anchor(name)})"
232def schema_type(schema: Any) -> str:
233 """A compact, human-readable type for one JSON Schema fragment.
235 Nullable unions (``anyOf`` with ``null``) read as ``string (nullable)``;
236 references link to the schema's section; free-form objects and empty
237 schemas are named as such rather than left blank.
238 """
239 if not isinstance(schema, dict):
240 return "any"
241 if "$ref" in schema:
242 return schema_link(_ref_name(str(schema["$ref"])))
243 for keyword, joiner in (("anyOf", " or "), ("oneOf", " or "), ("allOf", " and ")):
244 variants = schema.get(keyword)
245 if isinstance(variants, list) and variants:
246 non_null = [
247 v for v in variants if not (isinstance(v, dict) and v.get("type") == "null")
248 ]
249 nullable = len(non_null) < len(variants)
250 rendered = joiner.join(schema_type(v) for v in non_null) or "null"
251 return f"{rendered} (nullable)" if nullable else rendered
252 if "enum" in schema and isinstance(schema["enum"], list):
253 values = ", ".join(code(json.dumps(v)) for v in schema["enum"])
254 return f"enum: {values}"
255 if "const" in schema:
256 return f"const {code(json.dumps(schema['const']))}"
257 declared = schema.get("type")
258 types = declared if isinstance(declared, list) else [declared] if declared else []
259 if not types:
260 return (
261 "object (free-form)"
262 if "additionalProperties" in schema or "properties" in schema
263 else "any"
264 )
265 parts: list[str] = []
266 for type_name in types:
267 if type_name == "array":
268 parts.append(f"array of {schema_type(schema.get('items', {}))}")
269 elif type_name == "object":
270 additional = schema.get("additionalProperties")
271 if isinstance(additional, dict) and additional:
272 parts.append(f"object of {schema_type(additional)}")
273 elif additional is True or (additional == {} and "properties" not in schema):
274 parts.append("object (free-form)")
275 else:
276 parts.append("object")
277 elif type_name == "string" and schema.get("format"):
278 parts.append(f"string ({schema['format']})")
279 else:
280 parts.append(str(type_name))
281 return " or ".join(parts)
284_CONSTRAINT_KEYS: tuple[tuple[str, str], ...] = (
285 ("minimum", "≥ {}"),
286 ("exclusiveMinimum", "> {}"),
287 ("maximum", "≤ {}"),
288 ("exclusiveMaximum", "< {}"),
289 ("minLength", "min length {}"),
290 ("maxLength", "max length {}"),
291 ("minItems", "min items {}"),
292 ("maxItems", "max items {}"),
293 ("pattern", "pattern {}"),
294 ("multipleOf", "multiple of {}"),
295)
298def _constraint_fragments(schema: Any) -> list[str]:
299 if not isinstance(schema, dict):
300 return []
301 found = [
302 template.format(code(str(schema[key])) if key == "pattern" else schema[key])
303 for key, template in _CONSTRAINT_KEYS
304 if key in schema
305 ]
306 # FastAPI puts the bounds of an ``Optional[...]`` field on the non-null
307 # variant of its ``anyOf``, so a nullable ``str`` with ``max_length`` is
308 # only described if the variants are read too.
309 for keyword in ("anyOf", "oneOf", "allOf"):
310 variants = schema.get(keyword)
311 if isinstance(variants, list):
312 for variant in variants:
313 for fragment in _constraint_fragments(variant):
314 if fragment not in found:
315 found.append(fragment)
316 return found
319def constraints(schema: Any) -> str:
320 """Numeric/length/pattern bounds as ``≥ 1; ≤ 1000``, or a dash."""
321 return cell("; ".join(_constraint_fragments(schema)))
324def default_of(schema: Any) -> str:
325 if isinstance(schema, dict) and "default" in schema:
326 return code(json.dumps(schema["default"]))
327 return "—"
330def _description(*candidates: Any) -> str:
331 """The first non-empty description among the candidates, as one table cell."""
332 for candidate in candidates:
333 if isinstance(candidate, str) and candidate.strip():
334 return cell(candidate)
335 return "—"
338def _prose(text: Any) -> list[str]:
339 """A description as its own paragraph(s), trailing whitespace trimmed."""
340 if not isinstance(text, str) or not text.strip():
341 return []
342 return [line.rstrip() for line in text.strip().splitlines()]
345# ─── operations ──────────────────────────────────────────────────────────────
348def _operations(document: dict[str, Any]) -> list[tuple[str, str, dict[str, Any]]]:
349 """``(method, path, operation)`` in path order, methods in reading order."""
350 found: list[tuple[str, str, dict[str, Any]]] = []
351 for path, item in document["paths"].items():
352 if not isinstance(item, dict):
353 continue
354 for method in _METHOD_ORDER:
355 operation = item.get(method)
356 if isinstance(operation, dict):
357 found.append((method.upper(), str(path), operation))
358 return found
361def operation_heading(method: str, path: str) -> str:
362 return f"{method} {path}"
365def operation_anchor(method: str, path: str) -> str:
366 """``op-get-api-v1-jobs-namespace-name``: an explicit anchor for the operation.
368 Headings get an ``<a id>`` of our own rather than relying on the renderer's
369 heading slug: GitHub and python-markdown agree on ``GET /api/v1/jobs`` but
370 not on the root path ``GET /`` (GitHub keeps a trailing hyphen, Markdown
371 strips it), and an anchor both honour is worth one extra element.
372 """
373 tail = re.sub(r"[^a-z0-9]+", "-", path.lower()).strip("-")
374 return f"op-{method.lower()}" + (f"-{tail}" if tail else "")
377def schema_anchor(name: str) -> str:
378 return f"schema-{slugify(name)}"
381def anchored_heading(anchor: str, heading: str) -> list[str]:
382 return [f'<a id="{anchor}"></a>', "", f"### {code(heading)}", ""]
385def _content_summary(content: Any) -> str:
386 """``application/json: <type>`` for each media type of a body or response."""
387 if not isinstance(content, dict) or not content:
388 return "—"
389 return cell(
390 "; ".join(
391 f"{code(media_type)}: {schema_type(media.get('schema') if isinstance(media, dict) else None)}"
392 for media_type, media in content.items()
393 )
394 )
397def render_operation(method: str, path: str, operation: dict[str, Any]) -> list[str]:
398 lines = anchored_heading(operation_anchor(method, path), operation_heading(method, path))
399 lines.extend(_prose(operation.get("description") or operation.get("summary")))
400 if lines[-1] != "":
401 lines.append("")
402 facts = []
403 if operation.get("operationId"):
404 facts.append(f"- **Operation ID:** {code(str(operation['operationId']))}")
405 tags = operation.get("tags")
406 if isinstance(tags, list) and tags:
407 facts.append(f"- **Tags:** {', '.join(str(tag) for tag in tags)}")
408 if operation.get("deprecated") is True:
409 facts.append("- **Deprecated:** yes")
410 facts.extend(_security_facts(operation.get("security")))
411 facts.extend(_integration_facts(operation.get("x-amazon-apigateway-integration")))
412 facts.extend(_route_facts(operation.get("x-gco-route")))
413 if facts:
414 lines.extend([*facts, ""])
415 lines.extend(_backend_block(operation.get("x-gco-backend")))
417 parameters = operation.get("parameters")
418 if isinstance(parameters, list) and parameters:
419 rows = []
420 for parameter in parameters:
421 if not isinstance(parameter, dict):
422 continue
423 schema = parameter.get("schema")
424 rows.append(
425 [
426 code(str(parameter.get("name", ""))),
427 cell(parameter.get("in")),
428 schema_type(schema),
429 "yes" if parameter.get("required") is True else "no",
430 default_of(schema),
431 constraints(schema),
432 _description(
433 parameter.get("description"),
434 schema.get("description") if isinstance(schema, dict) else None,
435 ),
436 ]
437 )
438 lines.append("**Parameters**")
439 lines.append("")
440 lines.extend(
441 table(["Name", "In", "Type", "Required", "Default", "Constraints", "Description"], rows)
442 )
443 lines.append("")
445 body = operation.get("requestBody")
446 if isinstance(body, dict):
447 required = "required" if body.get("required") is True else "optional"
448 lines.append(f"**Request body** ({required}): {_content_summary(body.get('content'))}")
449 lines.append("")
451 responses = operation.get("responses")
452 if isinstance(responses, dict) and responses:
453 rows = [
454 [
455 code(str(status)),
456 _description(response.get("description") if isinstance(response, dict) else None),
457 _content_summary(response.get("content") if isinstance(response, dict) else None),
458 ]
459 for status, response in sorted(responses.items(), key=lambda item: str(item[0]))
460 ]
461 lines.append("**Responses**")
462 lines.append("")
463 lines.extend(table(["Status", "Description", "Content"], rows))
464 lines.append("")
465 return lines
468def _security_facts(security: Any) -> list[str]:
469 """``- **Security:** ...`` when the operation declares its requirements.
471 FastAPI exports omit the key (authentication is middleware, not schema);
472 the gateway documents write it explicitly, ``[]`` meaning unauthenticated.
473 """
474 if not isinstance(security, list):
475 return []
476 if not security:
477 return ["- **Security:** none (unauthenticated)"]
478 names = sorted(
479 {
480 str(name)
481 for requirement in security
482 if isinstance(requirement, dict)
483 for name in requirement
484 }
485 )
486 return [f"- **Security:** {', '.join(code(name) for name in names)}"]
489def _integration_facts(integration: Any) -> list[str]:
490 """API Gateway's integration type, timeout and transfer mode as one line."""
491 if not isinstance(integration, dict):
492 return []
493 parts = [code(str(integration.get("type", "?")))]
494 timeout = integration.get("timeoutInMillis")
495 if isinstance(timeout, int):
496 parts.append(f"{timeout / 1000:g} s integration timeout")
497 if integration.get("responseTransferMode") == "STREAM":
498 parts.append("response streaming")
499 return [f"- **Integration:** {', '.join(parts)}"]
502def _route_facts(route: Any) -> list[str]:
503 """The HTTPRoute rule that delivers an operation of the cluster gateway."""
504 if not isinstance(route, dict):
505 return []
506 backend = f"{route.get('service')}:{route.get('port')}"
507 return [
508 f"- **Delivered by rule:** {code(str(route.get('rule')))} → Service "
509 f"{code(backend)} in {code(str(route.get('namespace')))}"
510 ]
513def _backend_block(backend: Any) -> list[str]:
514 """The Lambda behind an API Gateway operation and the hops that follow it."""
515 if not isinstance(backend, dict):
516 return []
517 lines = ["**Backend**", ""]
518 function = backend.get("lambda")
519 if isinstance(function, dict):
520 detail = ", ".join(
521 str(function[key]) for key in ("runtime", "handler") if function.get(key) is not None
522 )
523 timeout = function.get("timeoutSeconds")
524 if isinstance(timeout, int):
525 detail += f", {timeout} s timeout"
526 lines.append(
527 f"- **Lambda:** {code(str(function.get('name')))} "
528 f"({code(str(function.get('source')))}; {detail})"
529 )
530 else:
531 lines.append(f"- **Integration:** {code(str(backend.get('integration')))}")
532 description = backend.get("description")
533 if isinstance(description, str) and description.strip():
534 lines.append(f"- **Behaviour:** {cell(description)}")
535 hops = backend.get("hops")
536 valid_hops = [hop for hop in hops if isinstance(hop, dict)] if isinstance(hops, list) else []
537 if valid_hops:
538 lines.append("- **Then:**")
539 for index, hop in enumerate(valid_hops, start=1):
540 documents = hop.get("documents")
541 links = (
542 " — " + ", ".join(f"[{code(str(name))}]({name}.md)" for name in documents)
543 if isinstance(documents, list) and documents
544 else ""
545 )
546 lines.append(f" {index}. {cell(hop.get('label'))}{links}")
547 lines.append("")
548 return lines
551# ─── document-level extensions ───────────────────────────────────────────────
554def _json_block(value: Any) -> list[str]:
555 return ["```json", json.dumps(value, indent=2, sort_keys=True), "```", ""]
558def render_servers(servers: Any) -> list[str]:
559 """``## Servers``: each URL template and its variables."""
560 if not isinstance(servers, list) or not servers:
561 return []
562 lines = ["## Servers", ""]
563 for server in servers:
564 if not isinstance(server, dict):
565 continue
566 lines.append(f"- {code(str(server.get('url')))}")
567 if server.get("description"):
568 lines.extend(["", *(f" {line}" for line in _prose(server.get("description")))])
569 variables = server.get("variables")
570 if isinstance(variables, dict) and variables:
571 lines.append("")
572 rows = [
573 [
574 code(str(name)),
575 code(json.dumps(spec.get("default")))
576 if isinstance(spec, dict) and "default" in spec
577 else "—",
578 _description(spec.get("description") if isinstance(spec, dict) else None),
579 ]
580 for name, spec in variables.items()
581 ]
582 lines.extend(f" {row}" for row in table(["Variable", "Default", "Description"], rows))
583 lines.append("")
584 return lines
587def render_security_schemes(schemes: Any) -> list[str]:
588 """``## Security schemes``: how a caller authenticates to this surface."""
589 if not isinstance(schemes, dict) or not schemes:
590 return []
591 lines = ["## Security schemes", ""]
592 for name, scheme in schemes.items():
593 if not isinstance(scheme, dict):
594 continue
595 where = ""
596 if scheme.get("in") and scheme.get("name"):
597 where = f" — {scheme['in']} {code(str(scheme['name']))}"
598 authtype = scheme.get("x-amazon-apigateway-authtype")
599 if authtype:
600 where += f" ({authtype})"
601 lines.append(f"### {code(str(name))}")
602 lines.append("")
603 lines.append(f"- **Type:** {cell(scheme.get('type'))}{where}")
604 if scheme.get("description"):
605 lines.extend(["", *_prose(scheme.get("description"))])
606 lines.append("")
607 return lines
610def render_gateway_details(document: dict[str, Any]) -> list[str]:
611 """``## Deployment``: the API Gateway facts the synthesized template carried."""
612 stage = document.get("x-gco-stage")
613 policy = document.get("x-gco-resource-policy")
614 if not isinstance(stage, dict) and not isinstance(policy, list):
615 return []
616 lines = ["## Deployment", ""]
617 endpoint_type = document.get("x-gco-endpoint-type")
618 if endpoint_type:
619 lines.append(f"- **Endpoint type:** {code(str(endpoint_type))}")
620 if isinstance(stage, dict):
621 throttling = _dict(stage.get("throttling"))
622 lines.append(
623 f"- **Stage:** {code(str(stage.get('name')))} — throttling "
624 f"{throttling.get('rateLimit')} req/s (burst {throttling.get('burstLimit')}), "
625 f"execution logging {code(str(stage.get('loggingLevel')))}, data trace "
626 f"{'on' if stage.get('dataTraceEnabled') else 'off'}, metrics "
627 f"{'on' if stage.get('metricsEnabled') else 'off'}, X-Ray tracing "
628 f"{'on' if stage.get('tracingEnabled') else 'off'}, access logs "
629 f"{'on' if stage.get('accessLogging') else 'off'}"
630 )
631 conditions = document.get("x-gco-conditions")
632 if isinstance(conditions, list) and conditions:
633 lines.append("- **Conditional routes:** operations marked *Only when* exist under:")
634 for condition in conditions:
635 if isinstance(condition, dict):
636 lines.append(
637 f" - {code(str(condition.get('id')))}: {cell(condition.get('description'))}"
638 )
639 lines.append("")
640 if isinstance(policy, list):
641 lines.extend(["### Resource policy", ""])
642 lines.append(
643 "Statements deployed on the REST API (`${AWS::AccountId}` is the deploying account):"
644 )
645 lines.append("")
646 lines.extend(_json_block(policy))
647 direct = document.get("x-gco-resource-policy-direct-access")
648 if isinstance(direct, list) and direct:
649 lines.append(
650 "Added when `api_gateway.regional_api_enabled` is true (and always outside "
651 "the `aws` partition), admitting the deploying account's other principals:"
652 )
653 lines.append("")
654 lines.extend(_json_block(direct))
655 waf = document.get("x-gco-waf")
656 if isinstance(waf, list) and waf:
657 lines.extend(["### WAF", ""])
658 rows = [
659 [
660 str(rule.get("priority")),
661 code(str(rule.get("name"))),
662 cell(rule.get("action")),
663 code(str(rule["managedRuleGroup"])) if rule.get("managedRuleGroup") else "—",
664 ]
665 for rule in waf
666 if isinstance(rule, dict)
667 ]
668 lines.extend(table(["Priority", "Rule", "Action", "Managed rule group"], rows))
669 lines.append("")
670 return lines
673def render_cluster_gateway_details(document: dict[str, Any]) -> list[str]:
674 """``## Routing``: the Gateway, its HTTPRoute table and what it leaves out."""
675 gateway = document.get("x-gco-gateway")
676 routes = document.get("x-gco-routes")
677 if not isinstance(gateway, dict) or not isinstance(routes, list):
678 return []
679 lines = ["## Routing", ""]
680 listener = _dict(gateway.get("listener"))
681 load_balancer = _dict(gateway.get("loadBalancer"))
682 qualified = f"{gateway.get('namespace')}/{gateway.get('name')}"
683 lines.append(
684 f"- **Gateway:** {code(qualified)} "
685 f"(GatewayClass {code(str(gateway.get('gatewayClassName')))}, controller "
686 f"{code(str(gateway.get('controllerName')))})"
687 )
688 lines.append(
689 f"- **Listener:** {code(str(listener.get('name')))} "
690 f"{listener.get('protocol')}:{listener.get('port')}; load balancer scheme "
691 f"{code(str(load_balancer.get('scheme')))}"
692 )
693 lines.append(f"- **HTTPRoute:** {code(str(gateway.get('httpRoute')))}")
694 unauthenticated = document.get("x-gco-unauthenticated-paths")
695 if isinstance(unauthenticated, list) and unauthenticated:
696 lines.append(
697 "- **Paths served without the HMAC envelope:** "
698 + ", ".join(code(str(path)) for path in unauthenticated)
699 )
700 lines.append("")
701 lines.extend(["### HTTPRoute rules", ""])
702 lines.append(
703 "In document order; the controller awards precedence by path specificity "
704 "(longest prefix wins), so `/` is the catch-all."
705 )
706 lines.append("")
707 rows = [
708 [code(str(rule.get("prefix"))), code(str(rule.get("service"))), str(rule.get("port"))]
709 for rule in routes
710 if isinstance(rule, dict)
711 ]
712 lines.extend(table(["Path prefix", "Service", "Port"], rows))
713 lines.append("")
714 for key, heading in (
715 ("loadBalancer", "Load balancer"),
716 ("targetGroup", "Target group"),
717 ):
718 value = gateway.get(key)
719 if isinstance(value, dict) and value:
720 lines.extend([f"### {heading}", "", *_json_block(value)])
721 unreachable = document.get("x-gco-unreachable")
722 if isinstance(unreachable, list) and unreachable:
723 lines.extend(["### Paths a Service serves but never receives", ""])
724 lines.append(
725 "A prefix cannot be split between Services, so these implementations are "
726 "reachable only by addressing the Service inside the cluster."
727 )
728 lines.append("")
729 rows = [
730 [
731 code(str(item.get("path"))),
732 code(str(item.get("service"))),
733 code(str(item.get("rule"))),
734 code(str(item.get("deliveredTo"))),
735 ]
736 for item in unreachable
737 if isinstance(item, dict)
738 ]
739 lines.extend(table(["Path", "Implemented by", "Winning rule", "Delivered to"], rows))
740 lines.append("")
741 not_exposed = document.get("x-gco-not-exposed")
742 if isinstance(not_exposed, dict) and not_exposed:
743 lines.extend(["### Services not on the ALB", ""])
744 for service, detail in not_exposed.items():
745 info = _dict(detail)
746 lines.append(
747 f"- [{code(str(service))}]({service}.md) — reached through "
748 f"[{code(str(info.get('via')))}]({info.get('via')}.md) "
749 f"({code(str(info.get('pathPrefix')) + '*')}). {cell(info.get('description'))}"
750 )
751 lines.append("")
752 downstream = document.get("x-gco-downstream")
753 if isinstance(downstream, dict) and downstream:
754 lines.extend(["### Downstream of the backends", ""])
755 for service, detail in downstream.items():
756 info = _dict(detail)
757 lines.append(
758 f"- [{code(str(service))}]({service}.md) → {cell(info.get('target'))}. "
759 f"{cell(info.get('description'))}"
760 )
761 lines.append("")
762 return lines
765# ─── schemas ─────────────────────────────────────────────────────────────────
768def render_schema(name: str, schema: dict[str, Any]) -> list[str]:
769 lines = anchored_heading(schema_anchor(name), name)
770 lines.extend(_prose(schema.get("description")))
771 if lines[-1] != "":
772 lines.append("")
773 properties = schema.get("properties")
774 required = schema.get("required")
775 required_names = {str(item) for item in required} if isinstance(required, list) else set()
776 if isinstance(properties, dict) and properties:
777 rows = [
778 [
779 code(str(prop_name)),
780 schema_type(prop),
781 "yes" if prop_name in required_names else "no",
782 default_of(prop),
783 constraints(prop),
784 _description(prop.get("description") if isinstance(prop, dict) else None),
785 ]
786 for prop_name, prop in properties.items()
787 ]
788 lines.extend(
789 table(["Property", "Type", "Required", "Default", "Constraints", "Description"], rows)
790 )
791 lines.append("")
792 else:
793 lines.append(f"- **Type:** {schema_type(schema)}")
794 lines.append("")
795 if "example" in schema:
796 lines.extend(
797 ["Example:", "", "```json", json.dumps(schema["example"], indent=2), "```", ""]
798 )
799 return lines
802# ─── the sheet ───────────────────────────────────────────────────────────────
805def banner(service: str) -> str:
806 return (
807 f"<!-- Generated by diagrams/api_specs/generate.py from docs/openapi/{service}.json. "
808 f"Do not edit by hand; regenerate with `{REGENERATION_COMMAND}`. -->"
809 )
812def _assert_unique_anchors(anchors: list[tuple[str, str]]) -> None:
813 """``(anchor, owner)`` pairs must not collide, or a link would land on the wrong section."""
814 seen: dict[str, str] = {}
815 for anchor, owner in anchors:
816 if anchor in seen:
817 raise SpecSheetError(
818 f"{seen[anchor]!r} and {owner!r} share the anchor #{anchor}; "
819 "the summary table could not link to both"
820 )
821 seen[anchor] = owner
824def render_sheet(service: str, document: dict[str, Any], *, site_url: str, repo_url: str) -> str:
825 info = _dict(document.get("info"))
826 title = str(info.get("title") or service)
827 operations = _operations(document)
828 schemas = _dict(_dict(document.get("components")).get("schemas"))
829 _assert_unique_anchors(
830 [
831 (operation_anchor(method, path), operation_heading(method, path))
832 for method, path, _ in operations
833 ]
834 + [(schema_anchor(str(name)), str(name)) for name in schemas]
835 )
837 noun, origin = _KINDS[document_kind(document)]
839 lines = [f"# {title} — API spec sheet", "", banner(service), ""]
840 lines.append(
841 f"*{noun} {code(service)} · OpenAPI {document.get('openapi', '?')} · "
842 f"API version {info.get('version', '?')} · {len(operations)} endpoints · {len(schemas)} schemas*"
843 )
844 lines.append("")
845 lines.extend(_prose(info.get("description")))
846 if lines[-1] != "":
847 lines.append("")
848 lines.extend(
849 [
850 f"- **Machine-readable document:** [{code(f'docs/openapi/{service}.json')}]"
851 f"({repo_url}/blob/main/docs/openapi/{service}.json) — the {origin}",
852 f"- **Interactive console (Swagger UI):** <{site_url}{SWAGGER_SITE_PREFIX}/{service}/>",
853 "- **Catalogue index:** [README.md](README.md) · "
854 "[interaction diagram](README.md#how-the-surfaces-fit-together)",
855 "",
856 ]
857 )
858 lines.extend(render_servers(document.get("servers")))
859 lines.extend(render_gateway_details(document))
860 lines.extend(render_cluster_gateway_details(document))
861 lines.extend(render_security_schemes(_dict(document.get("components")).get("securitySchemes")))
862 lines.extend(["## Endpoints", ""])
863 rows = []
864 for method, path, operation in operations:
865 anchor = operation_anchor(method, path)
866 tags = operation.get("tags")
867 rows.append(
868 [
869 f"[{code(method)}](#{anchor})",
870 code(path),
871 _description(operation.get("summary")),
872 cell(", ".join(str(tag) for tag in tags))
873 if isinstance(tags, list) and tags
874 else "—",
875 ]
876 )
877 lines.extend(table(["Method", "Path", "Summary", "Tags"], rows))
878 lines.extend(["", "## Endpoint details", ""])
879 for method, path, operation in operations:
880 lines.extend(render_operation(method, path, operation))
881 lines.extend(["## Schemas", ""])
882 if schemas:
883 for name, schema in schemas.items():
884 lines.extend(render_schema(str(name), schema if isinstance(schema, dict) else {}))
885 else:
886 lines.extend([f"This {noun.lower()} declares no component schemas.", ""])
887 return "\n".join(lines).rstrip() + "\n"
890def render_index(documents: dict[str, dict[str, Any]], *, site_url: str, repo_url: str) -> str:
891 rows = []
892 for service in catalogue_order(documents):
893 document = documents[service]
894 info = _dict(document.get("info"))
895 rows.append(
896 [
897 f"[{code(service)}]({service}.md)",
898 _KINDS[document_kind(document)][0],
899 cell(info.get("title")),
900 code(str(info.get("version", "?"))),
901 str(len(_operations(document))),
902 f"[{code('json')}]({repo_url}/blob/main/docs/openapi/{service}.json)",
903 f"[{code('/docs')}]({site_url}{SWAGGER_SITE_PREFIX}/{service}/)",
904 ]
905 )
906 lines = [
907 "# GCO API Spec Sheets",
908 "",
909 "<!-- Generated by diagrams/api_specs/generate.py from docs/openapi/*.json. "
910 f"Do not edit by hand; regenerate with `{REGENERATION_COMMAND}`. -->",
911 "",
912 "One spec sheet per HTTP surface of GCO, rendered from the OpenAPI document that",
913 "describes it: for each FastAPI **service** the document the application",
914 "generates itself (the one behind its automatic Swagger `/docs` page); for each",
915 "AWS **API Gateway** the document read out of the synthesized CDK stack; and for",
916 "the in-cluster **Gateway** (the internal ALB) the document composed from its",
917 "HTTPRoute and the service documents. The documents are committed under",
918 "`docs/openapi/`, each guarded by a `--check` in CI, and these sheets are a",
919 "deterministic rendering of them — so a route, parameter or model change shows up",
920 "here in the same pull request as the code.",
921 "",
922 *table(
923 ["Document", "Kind", "API", "Version", "Endpoints", "OpenAPI document", "Swagger UI"],
924 rows,
925 ),
926 "",
927 "The **Swagger UI** column is the Swagger console for each document (FastAPI's own",
928 f"`/docs` page for the services), served from the project site under `/{SWAGGER_SITE_PREFIX}/`",
929 "with a self-hosted copy of `swagger-ui-dist` (the site makes no third-party",
930 "requests). It is built at deploy time by `pages.yml`; the sheets in this directory",
931 "are also staged into the wiki's source tree under `/api/` by `scripts/build_wiki.py`,",
932 "so both renderings come from one source.",
933 "",
934 "## How the surfaces fit together",
935 "",
936 f"",
937 "",
938 "Drawn from the same documents: a client signs a request with SigV4 to one of the",
939 "two API Gateways; a Lambda proxy adds the request-bound HMAC envelope and forwards",
940 "it — over Global Accelerator from the global API, inside the VPC from the regional",
941 "bridge — to the region's internal ALB, the Kubernetes Gateway whose HTTPRoute picks",
942 "the Service by longest path prefix. The aggregator answers `/api/v1/global/*` by",
943 "fanning out to the regional API Gateways. Dashed arrows are opt-in or dynamic",
944 "paths; `†` marks routes that exist only under a deployment condition. Every box",
945 "is a document in the table above, and each is a link to its sheet on the site.",
946 "",
947 "## Regenerating",
948 "",
949 "```bash",
950 *_aligned_commands(
951 ("python scripts/generate_openapi.py", "refresh the service documents from the apps"),
952 (
953 "python scripts/generate_api_gateway_openapi.py",
954 "re-synthesize the API Gateway documents",
955 ),
956 (
957 "python scripts/generate_cluster_gateway_openapi.py",
958 "recompose the cluster gateway document",
959 ),
960 (REGENERATION_COMMAND, "rewrite these sheets, this index and the diagram"),
961 ("python diagrams/generate.py --check", "fail if anything here is stale"),
962 ),
963 "```",
964 "",
965 "To browse the consoles locally, install the locked npm tooling and serve the",
966 "generated tree:",
967 "",
968 "```bash",
969 "npm ci --ignore-scripts --no-audit --no-fund",
970 f"python diagrams/api_specs/generate.py --swagger-ui-dir /tmp/gco-{SWAGGER_SITE_PREFIX} \\",
971 " --swagger-assets node_modules/swagger-ui-dist",
972 f"python -m http.server --directory /tmp/gco-{SWAGGER_SITE_PREFIX} 8080",
973 "```",
974 "",
975 "Swagger UI loads the document with a browser request, so the tree has to be",
976 "served over HTTP rather than opened from the filesystem.",
977 ]
978 return "\n".join(lines) + "\n"
981# ─── the interaction diagram ─────────────────────────────────────────────────
982#
983# ``api-topology.svg`` is drawn from the documents alone: the API Gateway
984# documents' ``x-gco-backend`` hop chains, the cluster gateway's ``x-gco-routes``
985# (plus what it declares not exposed or downstream) and the service documents.
986# The layout is a fixed grid of columns — clients, API Gateways, Lambda
987# proxies, transit, the cluster Gateway, Services, what sits behind them — so
988# the picture is a deterministic function of the JSON and ``--check`` can
989# compare it byte for byte. Every box that is a document links to its sheet.
991_SVG_FONT = "ui-sans-serif, system-ui, -apple-system, 'Segoe UI', Helvetica, Arial, sans-serif"
992_SVG_MONO = "ui-monospace, SFMono-Regular, Menlo, Consolas, monospace"
993#: ``(fill, stroke)`` per node style.
994_NODE_STYLES: dict[str, tuple[str, str]] = {
995 "clients": ("#ffffff", "#334155"),
996 "api-gateway": ("#dbeafe", "#1d4ed8"),
997 "lambda": ("#fef3c7", "#b45309"),
998 "transit": ("#ede9fe", "#6d28d9"),
999 "cluster-gateway": ("#dcfce7", "#15803d"),
1000 "service": ("#f1f5f9", "#334155"),
1001 "dynamic": ("#ffffff", "#64748b"),
1002}
1003_LEGEND: tuple[tuple[str, str], ...] = (
1004 ("api-gateway", "AWS API Gateway (SigV4)"),
1005 ("lambda", "Lambda proxy (HMAC envelope)"),
1006 ("transit", "AWS transit"),
1007 ("cluster-gateway", "Kubernetes Gateway (internal ALB)"),
1008 ("service", "FastAPI service"),
1009 ("dynamic", "dynamic or internal-only"),
1010)
1011_COLUMN_WIDTHS: tuple[int, ...] = (150, 250, 225, 180, 265, 205, 215)
1012#: Space after each column; wider where edge labels (route prefixes) sit.
1013_COLUMN_GAPS: tuple[int, ...] = (130, 175, 60, 60, 140, 130, 0)
1014_MARGIN = 20
1015_NODE_GAP = 26
1016_TITLE_HEIGHT = 30
1017_LINE_HEIGHT = 15
1018_CHAR_WIDTH = 6.3
1019_EDGE_COLOUR = "#475569"
1020_FANOUT_COLOUR = "#6d28d9"
1023class _Node:
1024 def __init__(
1025 self,
1026 node_id: str,
1027 column: int,
1028 title: str,
1029 lines: list[str],
1030 style: str,
1031 href: str | None = None,
1032 ) -> None:
1033 self.id = node_id
1034 self.column = column
1035 self.title = title
1036 self.lines = lines
1037 self.style = style
1038 self.href = href
1039 self.x = 0
1040 self.y = 0
1041 self.width = _COLUMN_WIDTHS[column]
1042 self.height = _TITLE_HEIGHT + _LINE_HEIGHT * len(lines) + 10
1045class _Edge:
1046 def __init__(
1047 self,
1048 source: str,
1049 target: str,
1050 label: str = "",
1051 dashed: bool = False,
1052 colour: str = _EDGE_COLOUR,
1053 ) -> None:
1054 self.source = source
1055 self.target = target
1056 self.labels = [label] if label else []
1057 self.dashed = dashed
1058 self.colour = colour
1061def _wrap(text: str, width: int, *, indent: str = "") -> list[str]:
1062 """Break ``text`` into lines that fit a node of ``width`` pixels in the mono font."""
1063 import textwrap
1065 chars = max(12, int((width - 18) / _CHAR_WIDTH))
1066 return textwrap.wrap(str(text), width=chars, subsequent_indent=indent) or [""]
1069def _route_lines(document: dict[str, Any], width: int) -> list[str]:
1070 """One entry per path of an API Gateway document: its methods, then ``†`` if conditional."""
1071 lines: list[str] = []
1072 for path, item in document["paths"].items():
1073 if not isinstance(item, dict):
1074 continue
1075 methods = [m.upper() for m in _METHOD_ORDER if isinstance(item.get(m), dict)]
1076 conditional = any("x-gco-condition" in item[m.lower()] for m in methods)
1077 text = f"{' '.join(methods)} {path}{' †' if conditional else ''}"
1078 lines.extend(_wrap(text, width, indent=" "))
1079 return lines
1082def _short_title(hop_id: str) -> str:
1083 return " ".join(word.capitalize() for word in hop_id.split("-"))
1086def topology_model(
1087 documents: dict[str, dict[str, Any]], *, site_url: str
1088) -> tuple[list[_Node], list[_Edge]]:
1089 """Nodes and edges of the interaction diagram, derived from the documents."""
1090 kinds = {name: document_kind(document) for name, document in documents.items()}
1091 gateways = [n for n in catalogue_order(documents) if kinds[n] == "aws-api-gateway"]
1092 clusters = [n for n in catalogue_order(documents) if kinds[n] == "cluster-gateway"]
1093 services = [n for n in catalogue_order(documents) if kinds[n] == "fastapi"]
1094 nodes: dict[str, _Node] = {}
1095 edges: dict[tuple[str, str], _Edge] = {}
1097 def link(name: str) -> str:
1098 return f"{site_url}api/{name}/"
1100 def connect(
1101 source: str,
1102 target: str,
1103 label: str = "",
1104 *,
1105 dashed: bool = False,
1106 colour: str = _EDGE_COLOUR,
1107 ) -> None:
1108 for node_id in (source, target):
1109 if node_id not in nodes:
1110 raise SpecSheetError(f"the diagram references an unknown node {node_id!r}")
1111 edge = edges.setdefault(
1112 (source, target), _Edge(source, target, dashed=dashed, colour=colour)
1113 )
1114 if label and label not in edge.labels:
1115 edge.labels.append(label)
1117 nodes["clients"] = _Node(
1118 "clients",
1119 0,
1120 "Clients",
1121 ["AWS SigV4 (IAM)", "gco CLI, awscurl,", "boto3, curl --aws-sigv4"],
1122 "clients",
1123 )
1124 not_exposed: dict[str, dict[str, Any]] = {}
1125 for name in clusters:
1126 for service, info in _dict(documents[name].get("x-gco-not-exposed")).items():
1127 not_exposed[str(service)] = _dict(info)
1128 for name in services:
1129 document = documents[name]
1130 column = 6 if name in not_exposed else 5
1131 lines = [f"{len(_operations(document))} endpoints", f"Service {name}"]
1132 nodes[f"doc:{name}"] = _Node(
1133 f"doc:{name}",
1134 column,
1135 str(_dict(document.get("info")).get("title") or name),
1136 lines,
1137 "dynamic" if name in not_exposed else "service",
1138 link(name),
1139 )
1140 for name in gateways:
1141 document = documents[name]
1142 nodes[f"doc:{name}"] = _Node(
1143 f"doc:{name}",
1144 1,
1145 str(_dict(document.get("info")).get("title") or name),
1146 _route_lines(document, _COLUMN_WIDTHS[1]),
1147 "api-gateway",
1148 link(name),
1149 )
1150 for name in clusters:
1151 document = documents[name]
1152 gateway = _dict(document.get("x-gco-gateway"))
1153 routes = [r for r in document.get("x-gco-routes", []) if isinstance(r, dict)]
1154 lines = [
1155 f"Gateway {gateway.get('namespace')}/{gateway.get('name')}",
1156 f"HTTPRoute {gateway.get('httpRoute')} (prefix → Service)",
1157 *[f"{r.get('prefix')} → {r.get('service')}" for r in routes],
1158 ]
1159 nodes[f"doc:{name}"] = _Node(
1160 f"doc:{name}",
1161 4,
1162 str(_dict(document.get("info")).get("title") or name),
1163 [line for text in lines for line in _wrap(text, _COLUMN_WIDTHS[4], indent=" ")],
1164 "cluster-gateway",
1165 link(name),
1166 )
1168 # What the backends send on to (declared by the cluster gateway) is drawn
1169 # once, under the hop id the API Gateway chains use for the same place.
1170 for name in clusters:
1171 for service, detail in _dict(documents[name].get("x-gco-downstream")).items():
1172 info = _dict(detail)
1173 node_id = f"hop:{info.get('hop') or service}"
1174 nodes[node_id] = _Node(
1175 node_id,
1176 6,
1177 _short_title(str(info.get("hop") or service)) + "s",
1178 _wrap(str(info.get("target") or ""), _COLUMN_WIDTHS[6])[:4],
1179 "dynamic",
1180 )
1182 # Lambda nodes first, so a hop that names another API's proxy Lambda
1183 # (the aggregator's fan-out ends at the regional bridge's proxy) resolves.
1184 lambdas_by_name: dict[str, str] = {}
1185 for name in gateways:
1186 for _method, _path, operation in _operations(documents[name]):
1187 function = _dict(_dict(operation.get("x-gco-backend")).get("lambda"))
1188 if not function:
1189 continue
1190 node_id = f"lambda:{name}:{function.get('name')}"
1191 if node_id not in nodes:
1192 source = str(function.get("source") or "")
1193 nodes[node_id] = _Node(
1194 node_id,
1195 2,
1196 str(function.get("name")),
1197 [f"Lambda · {source.rsplit('/', 1)[-1]}", str(function.get("runtime") or "")],
1198 "lambda",
1199 )
1200 lambdas_by_name.setdefault(str(function.get("name")), node_id)
1202 def hop_targets(hop: dict[str, Any], previous: _Node) -> list[str]:
1203 named = hop.get("documents")
1204 if isinstance(named, list) and named:
1205 return [f"doc:{document}" for document in named]
1206 hop_id = str(hop.get("id"))
1207 if hop_id in lambdas_by_name:
1208 return [lambdas_by_name[hop_id]]
1209 node_id = f"hop:{hop_id}"
1210 if node_id not in nodes:
1211 column = max(3, previous.column + 1)
1212 nodes[node_id] = _Node(
1213 node_id,
1214 column,
1215 _short_title(hop_id),
1216 _wrap(str(hop.get("label") or ""), _COLUMN_WIDTHS[column])[:4],
1217 "dynamic" if column >= 5 else "transit",
1218 )
1219 return [node_id]
1221 for name in gateways:
1222 document = documents[name]
1223 direct = bool(document.get("x-gco-resource-policy-direct-access"))
1224 connect("clients", f"doc:{name}", "SigV4 · opt-in" if direct else "SigV4", dashed=direct)
1225 # One label per path into a Lambda; a path is marked conditional when
1226 # any of its methods is.
1227 conditional_paths = {
1228 path for _m, path, operation in _operations(document) if "x-gco-condition" in operation
1229 }
1230 for _method, path, operation in _operations(document):
1231 backend = _dict(operation.get("x-gco-backend"))
1232 function = _dict(backend.get("lambda"))
1233 if not function:
1234 continue
1235 previous_id = f"lambda:{name}:{function.get('name')}"
1236 marker = " †" if path in conditional_paths else ""
1237 connect(f"doc:{name}", previous_id, f"{path}{marker}")
1238 for hop in backend.get("hops", []):
1239 if not isinstance(hop, dict):
1240 continue
1241 targets = hop_targets(hop, nodes[previous_id])
1242 fan_out = str(hop.get("id")) in {n.removeprefix("doc:") for n in targets} and any(
1243 kinds.get(t.removeprefix("doc:")) == "aws-api-gateway" for t in targets
1244 )
1245 for target in targets:
1246 connect(
1247 previous_id,
1248 target,
1249 "SigV4 fan-out" if fan_out else "",
1250 dashed=fan_out,
1251 colour=_FANOUT_COLOUR if fan_out else _EDGE_COLOUR,
1252 )
1253 if len(targets) != 1:
1254 break
1255 previous_id = targets[0]
1257 for name in clusters:
1258 document = documents[name]
1259 by_service: dict[str, list[str]] = {}
1260 for rule in document.get("x-gco-routes", []):
1261 if isinstance(rule, dict):
1262 by_service.setdefault(str(rule.get("service")), []).append(str(rule.get("prefix")))
1263 for service, prefixes in by_service.items():
1264 for prefix in prefixes:
1265 connect(f"doc:{name}", f"doc:{service}", prefix)
1266 for service, info in not_exposed.items():
1267 connect(
1268 f"doc:{info.get('via')}",
1269 f"doc:{service}",
1270 f"{info.get('pathPrefix')}*",
1271 dashed=True,
1272 )
1273 for service, detail in _dict(document.get("x-gco-downstream")).items():
1274 info = _dict(detail)
1275 connect(
1276 f"doc:{service}",
1277 f"hop:{info.get('hop') or service}",
1278 "by endpoint name",
1279 dashed=True,
1280 )
1282 return list(nodes.values()), list(edges.values())
1285def _label_lines(labels: list[str]) -> list[str]:
1286 """Edge labels as lines: paths under one deep prefix collapse to ``prefix*``."""
1287 if len(labels) > 1 and all(label.startswith("/") for label in labels):
1288 import os.path
1290 common = os.path.commonprefix(labels)
1291 prefix = common[: common.rfind("/") + 1]
1292 if prefix.count("/") >= 3:
1293 return [f"{prefix}*"]
1294 return labels
1297def _layout(nodes: list[_Node]) -> tuple[int, int]:
1298 """Place nodes on the column grid, columns vertically centred; return the canvas size."""
1299 columns: dict[int, list[_Node]] = {}
1300 for node in nodes:
1301 columns.setdefault(node.column, []).append(node)
1302 heights = {
1303 column: sum(n.height for n in members) + _NODE_GAP * (len(members) - 1)
1304 for column, members in columns.items()
1305 }
1306 total_height = max(heights.values()) if heights else 0
1307 x = _MARGIN
1308 for column, width in enumerate(_COLUMN_WIDTHS):
1309 y = _MARGIN + (total_height - heights.get(column, 0)) // 2
1310 for node in columns.get(column, []):
1311 node.x, node.y = x, y
1312 y += node.height + _NODE_GAP
1313 x += width + _COLUMN_GAPS[column]
1314 return x + _MARGIN, total_height + 2 * _MARGIN
1317def _svg_text(
1318 x: float,
1319 y: float,
1320 text: str,
1321 *,
1322 size: float,
1323 mono: bool = False,
1324 weight: str = "normal",
1325 anchor: str = "start",
1326 fill: str = "#0f172a",
1327) -> str:
1328 font = _SVG_MONO if mono else _SVG_FONT
1329 return (
1330 f'<text x="{x:g}" y="{y:g}" font-family="{font}" font-size="{size:g}" '
1331 f'font-weight="{weight}" text-anchor="{anchor}" fill="{fill}">{_xml(text)}</text>'
1332 )
1335def _xml(text: str) -> str:
1336 """Escape what would end or alter an SVG text node or attribute value."""
1337 return (
1338 text.replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """)
1339 )
1342def _svg_node(node: _Node) -> str:
1343 fill, stroke = _NODE_STYLES[node.style]
1344 dash = ' stroke-dasharray="6 4"' if node.style == "dynamic" else ""
1345 parts = [
1346 f'<rect x="{node.x}" y="{node.y}" width="{node.width}" height="{node.height}" rx="8" '
1347 f'fill="{fill}" stroke="{stroke}" stroke-width="1.5"{dash}/>',
1348 _svg_text(node.x + 10, node.y + 19, node.title, size=12.5, weight="bold", fill=stroke),
1349 ]
1350 for index, line in enumerate(node.lines):
1351 parts.append(
1352 _svg_text(
1353 node.x + 10,
1354 node.y + _TITLE_HEIGHT + 8 + _LINE_HEIGHT * index,
1355 line,
1356 size=10.5,
1357 mono=True,
1358 fill="#1e293b",
1359 )
1360 )
1361 body = "\n ".join(parts)
1362 if node.href:
1363 return f'<a href="{_xml(node.href)}"><title>{_xml(node.title)}</title>\n {body}\n </a>'
1364 return body
1367def _svg_edge(edge: _Edge, nodes: dict[str, _Node]) -> str:
1368 source, target = nodes[edge.source], nodes[edge.target]
1369 ty = target.y + target.height / 2
1370 forward = target.column > source.column
1371 if forward:
1372 # Right edge of the source, mid-height, to the left edge of the target.
1373 sx, sy, tx = source.x + source.width, source.y + source.height / 2, target.x
1374 mx = sx + 24
1375 else:
1376 # A hop back to an earlier column leaves from the source's lower left
1377 # so it never overprints the edge arriving at the source's midpoint.
1378 sx, sy, tx = source.x, source.y + source.height - 12, target.x + target.width
1379 mx = sx - 24
1380 path = f"M {sx:g} {sy:g} H {mx:g} V {ty:g} H {tx:g}"
1381 dash = ' stroke-dasharray="7 5"' if edge.dashed else ""
1382 marker = "arrow-fanout" if edge.colour == _FANOUT_COLOUR else "arrow"
1383 parts = [
1384 f'<path d="{path}" fill="none" stroke="{edge.colour}" stroke-width="1.4"{dash} '
1385 f'marker-end="url(#{marker})"/>'
1386 ]
1387 # Forward labels sit above the last segment, right-aligned at the target and
1388 # stacked upward one line per label so several prefixes into one Service
1389 # never overprint; a backward label hangs below its first segment.
1390 lx, base_y, step = (tx - 8, ty - 6, -12) if forward else (sx - 8, sy + 14, 12)
1391 for index, label in enumerate(reversed(_label_lines(edge.labels))):
1392 parts.append(
1393 _svg_text(
1394 lx,
1395 base_y + step * index,
1396 label,
1397 size=9.5,
1398 mono=True,
1399 anchor="end",
1400 fill=edge.colour,
1401 )
1402 )
1403 return "\n ".join(parts)
1406def render_topology(documents: dict[str, dict[str, Any]], *, site_url: str) -> str:
1407 """The interaction diagram as a standalone SVG document."""
1408 nodes, edges = topology_model(documents, site_url=site_url)
1409 by_id = {node.id: node for node in nodes}
1410 width, height = _layout(nodes)
1411 legend_y = height + 4
1412 legend_parts = []
1413 x = _MARGIN
1414 for style, label in _LEGEND:
1415 fill, stroke = _NODE_STYLES[style]
1416 dash = ' stroke-dasharray="4 3"' if style == "dynamic" else ""
1417 legend_parts.append(
1418 f'<rect x="{x}" y="{legend_y}" width="16" height="12" rx="3" fill="{fill}" '
1419 f'stroke="{stroke}" stroke-width="1.2"{dash}/>'
1420 )
1421 legend_parts.append(_svg_text(x + 22, legend_y + 10, label, size=10.5, fill="#334155"))
1422 x += 22 + int(len(label) * 6.0) + 26
1423 legend_parts.append(
1424 _svg_text(
1425 _MARGIN,
1426 legend_y + 30,
1427 "† route exists only under a deployment condition (see the sheet) · dashed arrow: "
1428 "opt-in or dynamic path · every box links to its spec sheet",
1429 size=10.5,
1430 fill="#334155",
1431 )
1432 )
1433 legend_parts.append(
1434 _svg_text(
1435 _MARGIN,
1436 legend_y + 48,
1437 f"Generated by diagrams/api_specs/generate.py from docs/openapi/*.json; "
1438 f"regenerate with `{REGENERATION_COMMAND}`.",
1439 size=9.5,
1440 fill="#64748b",
1441 )
1442 )
1443 total_height = legend_y + 60
1444 lines = [
1445 '<?xml version="1.0" encoding="UTF-8"?>',
1446 # The default SVG namespace only: links use SVG 2's plain ``href``, and
1447 # .github/scripts/validate_svg_assets.py refuses ``xlink`` (or any other
1448 # foreign namespace) in a tracked SVG.
1449 f'<svg xmlns="http://www.w3.org/2000/svg" '
1450 f'width="{width}" height="{total_height}" viewBox="0 0 {width} {total_height}" '
1451 'role="img" aria-labelledby="title desc">',
1452 ' <title id="title">GCO API topology: API Gateways, Lambda proxies, the cluster Gateway and the services</title>',
1453 ' <desc id="desc">Clients sign requests with SigV4 to the global or regional API Gateway; Lambda '
1454 "proxies add the HMAC envelope and forward to the region's internal ALB, the Kubernetes "
1455 "Gateway whose HTTPRoute routes by path prefix to the FastAPI services.</desc>",
1456 " <defs>",
1457 f' <marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="8" markerHeight="8" orient="auto-start-reverse"><path d="M 0 0 L 10 5 L 0 10 z" fill="{_EDGE_COLOUR}"/></marker>',
1458 f' <marker id="arrow-fanout" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="8" markerHeight="8" orient="auto-start-reverse"><path d="M 0 0 L 10 5 L 0 10 z" fill="{_FANOUT_COLOUR}"/></marker>',
1459 " </defs>",
1460 f' <rect width="{width}" height="{total_height}" fill="#ffffff"/>',
1461 ]
1462 lines.extend(f" {_svg_edge(edge, by_id)}" for edge in edges)
1463 lines.extend(f" {_svg_node(node)}" for node in nodes)
1464 lines.extend(f" {part}" for part in legend_parts)
1465 lines.append("</svg>")
1466 return "\n".join(lines) + "\n"
1469# ─── the catalogue contract ──────────────────────────────────────────────────
1472def expected_outputs(project_root: Path) -> dict[Path, str]:
1473 """``{path: content}`` for every file this catalogue owns, rendered in memory."""
1474 openapi_dir = project_root / "docs" / "openapi"
1475 output_dir = project_root / "diagrams" / "api_specs"
1476 site_url, repo_url = read_site_config(project_root / SITE_CONFIG_FILE)
1477 documents = {
1478 service: load_document(service, openapi_dir) for service in service_names(openapi_dir)
1479 }
1480 outputs = {
1481 output_dir / f"{service}.md": render_sheet(
1482 service, document, site_url=site_url, repo_url=repo_url
1483 )
1484 for service, document in documents.items()
1485 }
1486 outputs[output_dir / "README.md"] = render_index(
1487 documents, site_url=site_url, repo_url=repo_url
1488 )
1489 outputs[output_dir / TOPOLOGY_FILE] = render_topology(documents, site_url=site_url)
1490 return outputs
1493def api_contract_issues(project_root: Path) -> list[str]:
1494 """Stale, missing and orphan sheets, by repository path; empty when current.
1496 A checkout that cannot be rendered at all (no OpenAPI documents, no
1497 ``zensical.toml``, a colliding anchor) is reported as an issue rather than
1498 raised, so the aggregate ``--check`` prints one actionable line per
1499 catalogue instead of a traceback.
1500 """
1501 output_dir = project_root / "diagrams" / "api_specs"
1502 try:
1503 expected = expected_outputs(project_root)
1504 except (SpecSheetError, OSError) as exc:
1505 return [f"API spec sheets cannot be rendered: {exc}"]
1506 issues: list[str] = []
1507 for path, content in sorted(expected.items()):
1508 relative = path.relative_to(project_root).as_posix()
1509 if not path.is_file():
1510 issues.append(f"missing API spec sheet: {relative}")
1511 elif path.read_text(encoding="utf-8") != content:
1512 issues.append(f"stale API spec sheet: {relative}")
1513 for path in sorted([*output_dir.glob("*.md"), *output_dir.glob("*.svg")]):
1514 if path not in expected:
1515 issues.append(f"orphan API spec sheet: {path.relative_to(project_root).as_posix()}")
1516 if issues:
1517 issues.append(f"regenerate with `{REGENERATION_COMMAND}`")
1518 return issues
1521def write_outputs(project_root: Path) -> list[str]:
1522 """Write every owned file; return the repository paths that changed."""
1523 changed: list[str] = []
1524 for path, content in sorted(expected_outputs(project_root).items()):
1525 current = path.read_text(encoding="utf-8") if path.is_file() else None
1526 if current != content:
1527 path.parent.mkdir(parents=True, exist_ok=True)
1528 path.write_text(content, encoding="utf-8")
1529 changed.append(path.relative_to(project_root).as_posix())
1530 return changed
1533# ─── Swagger UI consoles ─────────────────────────────────────────────────────
1536def _console_index(services: dict[str, str], site_url: str) -> str:
1537 items = "\n".join(
1538 f' <li><a href="./{service}/">{title}</a> <code>{service}</code></li>'
1539 for service, title in sorted(services.items())
1540 )
1541 return f"""<!DOCTYPE html>
1542<html lang="en">
1543<head>
1544 <meta charset="utf-8">
1545 <meta name="viewport" content="width=device-width, initial-scale=1">
1546 <title>GCO API consoles</title>
1547 <link rel="icon" href="./assets/favicon-32x32.png">
1548 <style>
1549 body {{ font-family: system-ui, sans-serif; margin: 2rem auto; max-width: 42rem; line-height: 1.5; }}
1550 code {{ background: #f2f2f2; padding: 0 .3em; border-radius: 3px; }}
1551 </style>
1552</head>
1553<body>
1554 <h1>GCO API consoles</h1>
1555 <p>FastAPI's Swagger UI for each HTTP service, rendered from the committed OpenAPI
1556 documents. Read-only: these pages describe the API and do not send requests.</p>
1557 <ul>
1558{items}
1559 </ul>
1560 <p><a href="{site_url}">Back to the wiki</a></p>
1561</body>
1562</html>
1563"""
1566def build_swagger_site(site_dir: Path, assets_dir: Path, *, project_root: Path) -> list[Path]:
1567 """Write FastAPI's Swagger UI console per service under ``site_dir``.
1569 ``assets_dir`` is the ``swagger-ui-dist`` package directory; its bundle,
1570 stylesheet and favicon (plus LICENSE/NOTICE when shipped) are copied to
1571 ``site_dir/assets`` so the console loads everything from the site itself.
1572 Every generated page is checked to reference no remote resource, so a
1573 FastAPI default creeping back in fails the build instead of the site's
1574 zero-third-party-requests rule. Returns the written files.
1575 """
1576 from fastapi.openapi.docs import get_swagger_ui_html
1578 missing = [name for name in SWAGGER_ASSET_FILES if not (assets_dir / name).is_file()]
1579 if missing:
1580 raise SpecSheetError(
1581 f"swagger-ui-dist assets missing from {assets_dir}: {', '.join(missing)} "
1582 "(run `npm ci --ignore-scripts --no-audit --no-fund`)"
1583 )
1584 site_url, _repo_url = read_site_config(project_root / SITE_CONFIG_FILE)
1585 openapi_dir = project_root / "docs" / "openapi"
1586 written: list[Path] = []
1587 assets_out = site_dir / "assets"
1588 assets_out.mkdir(parents=True, exist_ok=True)
1589 for name in (*SWAGGER_ASSET_FILES, *SWAGGER_NOTICE_FILES):
1590 source = assets_dir / name
1591 if source.is_file():
1592 written.append(Path(shutil.copyfile(source, assets_out / name)))
1594 titles: dict[str, str] = {}
1595 for service in service_names(openapi_dir):
1596 document = load_document(service, openapi_dir)
1597 titles[service] = str(_dict(document.get("info")).get("title") or service)
1598 console_dir = site_dir / service
1599 console_dir.mkdir(parents=True, exist_ok=True)
1600 spec_copy = console_dir / "openapi.json"
1601 spec_copy.write_text(
1602 (openapi_dir / f"{service}.json").read_text(encoding="utf-8"), encoding="utf-8"
1603 )
1604 written.append(spec_copy)
1605 page = get_swagger_ui_html(
1606 openapi_url="openapi.json",
1607 title=f"{titles[service]} — Swagger UI",
1608 swagger_js_url="../assets/swagger-ui-bundle.js",
1609 swagger_css_url="../assets/swagger-ui.css",
1610 swagger_favicon_url="../assets/favicon-32x32.png",
1611 # No validator badge (a request to validator.swagger.io) and no
1612 # "Try it out": the site has no backend to send requests to.
1613 swagger_ui_parameters={"validatorUrl": None, "supportedSubmitMethods": []},
1614 )
1615 html = bytes(page.body).decode("utf-8")
1616 if re.search(r"(?:src|href)=\"https?://", html):
1617 raise SpecSheetError(f"{service}: the Swagger UI page references a remote resource")
1618 # FastAPI serves this page with a ``charset=utf-8`` header and so omits
1619 # the meta tag; a static host may not. Without it a browser falls back
1620 # to Latin-1 for the page *and* the bundle it loads, and the bundle's
1621 # non-ASCII literals then fail to parse ("SwaggerUIBundle is not defined").
1622 html = html.replace("<head>", '<head>\n <meta charset="utf-8">', 1)
1623 if '<meta charset="utf-8">' not in html:
1624 raise SpecSheetError(f"{service}: FastAPI's Swagger UI template has no <head> to tag")
1625 index = console_dir / "index.html"
1626 index.write_text(html, encoding="utf-8")
1627 written.append(index)
1628 root_index = site_dir / "index.html"
1629 root_index.write_text(_console_index(titles, site_url), encoding="utf-8")
1630 written.append(root_index)
1631 return written
1634# ─── CLI ─────────────────────────────────────────────────────────────────────
1637def main(argv: list[str] | None = None) -> int:
1638 parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
1639 parser.add_argument(
1640 "--check",
1641 action="store_true",
1642 help="Do not write; exit non-zero if any committed sheet is stale.",
1643 )
1644 parser.add_argument(
1645 "--swagger-ui-dir",
1646 type=Path,
1647 help="Also build the Swagger UI consoles into this directory (the Pages /swagger/ tree).",
1648 )
1649 parser.add_argument(
1650 "--swagger-assets",
1651 type=Path,
1652 default=None,
1653 help="The swagger-ui-dist package directory to self-host from "
1654 "(default: node_modules/swagger-ui-dist).",
1655 )
1656 args = parser.parse_args(argv)
1657 swagger_assets = args.swagger_assets or REPO_ROOT / "node_modules" / "swagger-ui-dist"
1659 try:
1660 if args.check:
1661 issues = api_contract_issues(REPO_ROOT)
1662 for issue in issues:
1663 print(f"ERROR: {issue}", file=sys.stderr)
1664 if issues:
1665 return 1
1666 print("API spec sheets are current")
1667 else:
1668 changed = write_outputs(REPO_ROOT)
1669 for path in changed:
1670 print(f"{path}: written")
1671 if not changed:
1672 print("API spec sheets unchanged")
1673 if args.swagger_ui_dir is not None:
1674 written = build_swagger_site(
1675 args.swagger_ui_dir, swagger_assets, project_root=REPO_ROOT
1676 )
1677 print(f"Swagger UI consoles: {len(written)} files under {args.swagger_ui_dir}")
1678 except SpecSheetError as exc:
1679 print(f"ERROR: {exc}", file=sys.stderr)
1680 return 2
1681 return 0
1684if __name__ == "__main__":
1685 raise SystemExit(main())