Wire cli/docs.py build/serve/generate commands to run extraction scripts and Docusaurus. Add extract_cli.py (27 CLI docs from typer help), extract_openapi.py (8 REST API docs from FastAPI schema), extract_citations.py (14 pipeline pages with bib references), extract_dag.py (pipeline + table lineage Mermaid diagrams). Closes #26, #27, #28, #29, #30.
172 lines
5.4 KiB
Python
172 lines
5.4 KiB
Python
"""Extract REST API docs from FastAPI OpenAPI schema.
|
|
|
|
Reads the OpenAPI JSON from the FastAPI app and generates Markdown
|
|
files for each route group into ``docs/docs/rest-api/``.
|
|
|
|
Usage::
|
|
|
|
uv run python docs/scripts/extract_openapi.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path("src").resolve()))
|
|
|
|
OUT_DIR = Path("docs/docs/rest-api")
|
|
|
|
|
|
def _type_str(schema: dict) -> str:
|
|
"""Convert a JSON Schema type to a readable string."""
|
|
if "$ref" in schema:
|
|
return schema["$ref"].rsplit("/", 1)[-1]
|
|
if "anyOf" in schema:
|
|
parts = [_type_str(s) for s in schema["anyOf"]]
|
|
return " | ".join(parts)
|
|
t = schema.get("type", "object")
|
|
if t == "array":
|
|
items = schema.get("items", {})
|
|
return f"list[{_type_str(items)}]"
|
|
return t
|
|
|
|
|
|
def _render_endpoint(method: str, path: str, spec: dict) -> str:
|
|
"""Render a single endpoint to markdown."""
|
|
lines = []
|
|
summary = spec.get("summary", "")
|
|
lines.append(f"### `{method.upper()} {path}`\n")
|
|
if summary:
|
|
lines.append(f"{summary}\n")
|
|
|
|
# Parameters
|
|
params = spec.get("parameters", [])
|
|
if params:
|
|
lines.append("**Parameters:**\n")
|
|
lines.append("| Name | In | Type | Required |")
|
|
lines.append("|------|-----|------|----------|")
|
|
for p in params:
|
|
schema = p.get("schema", {})
|
|
lines.append(
|
|
f"| `{p['name']}` | {p.get('in', '')} "
|
|
f"| {_type_str(schema)} | {p.get('required', False)} |"
|
|
)
|
|
lines.append("")
|
|
|
|
# Request body
|
|
body = spec.get("requestBody", {})
|
|
if body:
|
|
content = body.get("content", {})
|
|
for ct, ct_spec in content.items():
|
|
schema = ct_spec.get("schema", {})
|
|
lines.append(f"**Request body** (`{ct}`):\n")
|
|
lines.append(f"Type: `{_type_str(schema)}`\n")
|
|
|
|
# Responses
|
|
responses = spec.get("responses", {})
|
|
if responses:
|
|
lines.append("**Responses:**\n")
|
|
lines.append("| Status | Description |")
|
|
lines.append("|--------|-------------|")
|
|
for code, resp in sorted(responses.items()):
|
|
desc = resp.get("description", "")
|
|
lines.append(f"| {code} | {desc} |")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main() -> None:
|
|
from api.server import app # noqa: E402
|
|
|
|
schema = app.openapi()
|
|
|
|
if OUT_DIR.exists():
|
|
import shutil
|
|
|
|
shutil.rmtree(OUT_DIR)
|
|
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Category metadata
|
|
cat_json = OUT_DIR / "_category_.json"
|
|
cat_json.write_text('{"label": "REST API", "position": 11}\n')
|
|
|
|
# Write full OpenAPI JSON
|
|
(OUT_DIR / "openapi.json").write_text(json.dumps(schema, indent=2))
|
|
|
|
# Group endpoints by first tag
|
|
groups: dict[str, list[tuple[str, str, dict]]] = {}
|
|
for path, methods in schema.get("paths", {}).items():
|
|
for method, spec in methods.items():
|
|
if method in ("parameters", "servers", "summary"):
|
|
continue
|
|
tags = spec.get("tags", ["default"])
|
|
tag = tags[0] if tags else "default"
|
|
groups.setdefault(tag, []).append((method, path, spec))
|
|
|
|
count = 0
|
|
# Overview page
|
|
overview_lines = [
|
|
"---\ntitle: REST API Overview\nsidebar_position: 1\n---\n",
|
|
"# REST API\n",
|
|
f"**Title:** {schema.get('info', {}).get('title', 'Stack API')}\n",
|
|
f"**Version:** {schema.get('info', {}).get('version', '0.1.0')}\n",
|
|
"## Endpoints\n",
|
|
]
|
|
for tag, endpoints in sorted(groups.items()):
|
|
overview_lines.append(f"- [{tag}](./{tag}) ({len(endpoints)} endpoints)")
|
|
overview_lines.append(
|
|
"\n\nFull OpenAPI spec: [openapi.json](./openapi.json)\n"
|
|
)
|
|
(OUT_DIR / "index.md").write_text("\n".join(overview_lines))
|
|
count += 1
|
|
|
|
# Per-group pages
|
|
for pos, (tag, endpoints) in enumerate(sorted(groups.items()), start=2):
|
|
lines = [
|
|
f"---\ntitle: {tag}\nsidebar_position: {pos}\n---\n",
|
|
f"# {tag}\n",
|
|
]
|
|
for method, path, spec in endpoints:
|
|
lines.append(_render_endpoint(method, path, spec))
|
|
|
|
out_path = OUT_DIR / f"{tag}.md"
|
|
out_path.write_text("\n".join(lines))
|
|
count += 1
|
|
print(f" {tag} -> {out_path}")
|
|
|
|
# Schemas section
|
|
schemas = schema.get("components", {}).get("schemas", {})
|
|
if schemas:
|
|
schema_lines = [
|
|
"---\ntitle: Schemas\nsidebar_position: 99\n---\n",
|
|
"# Schemas\n",
|
|
]
|
|
for name, s in sorted(schemas.items()):
|
|
schema_lines.append(f"## `{name}`\n")
|
|
desc = s.get("description", "")
|
|
if desc:
|
|
schema_lines.append(f"{desc}\n")
|
|
props = s.get("properties", {})
|
|
if props:
|
|
schema_lines.append("| Field | Type | Required |")
|
|
schema_lines.append("|-------|------|----------|")
|
|
required = set(s.get("required", []))
|
|
for field, fspec in props.items():
|
|
schema_lines.append(
|
|
f"| `{field}` | {_type_str(fspec)} "
|
|
f"| {'yes' if field in required else ''} |"
|
|
)
|
|
schema_lines.append("")
|
|
|
|
(OUT_DIR / "schemas.md").write_text("\n".join(schema_lines))
|
|
count += 1
|
|
|
|
print(f"\nDone: {count} REST API docs written to {OUT_DIR}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|