Task 002: add shared domain contracts

scripts/generate_openapi_contracts.py

@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+import json
+import keyword
+import sys
+from pathlib import Path
+from typing import Any
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+BACKEND_ROOT = REPO_ROOT / "apps" / "backend"
+OPENAPI_PATH = REPO_ROOT / "packages" / "shared" / "openapi.json"
+TS_TYPES_PATH = REPO_ROOT / "packages" / "shared" / "src" / "api-types.ts"
+
+sys.path.insert(0, str(BACKEND_ROOT))
+
+from src.presentation.main import app  # noqa: E402
+
+
+GENERATED_HEADER = """// Generated from backend OpenAPI by scripts/generate_openapi_contracts.py.
+// Do not edit by hand.
+
+"""
+
+
+def main() -> None:
+    openapi = app.openapi()
+    OPENAPI_PATH.write_text(
+        json.dumps(openapi, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    TS_TYPES_PATH.write_text(render_typescript(openapi), encoding="utf-8")
+    print(f"wrote {OPENAPI_PATH}")
+    print(f"wrote {TS_TYPES_PATH}")
+
+
+def render_typescript(openapi: dict[str, Any]) -> str:
+    schemas = openapi.get("components", {}).get("schemas", {})
+    lines = [GENERATED_HEADER.rstrip(), ""]
+    for name in sorted(schemas):
+        lines.append(render_schema_type(name, schemas[name], schemas))
+        lines.append("")
+    return "\n".join(lines).rstrip() + "\n"
+
+
+def render_schema_type(
+    name: str,
+    schema: dict[str, Any],
+    schemas: dict[str, Any],
+) -> str:
+    if schema.get("type") == "object" or "properties" in schema:
+        return f"export type {name} = {render_object(schema, schemas, indent=0)};"
+    return f"export type {name} = {schema_to_ts(schema, schemas)};"
+
+
+def render_object(
+    schema: dict[str, Any],
+    schemas: dict[str, Any],
+    *,
+    indent: int,
+) -> str:
+    properties = schema.get("properties", {})
+    required = set(schema.get("required", []))
+    pad = "  " * indent
+    child_pad = "  " * (indent + 1)
+
+    if not properties:
+        additional = schema.get("additionalProperties")
+        if isinstance(additional, dict):
+            return f"Record<string, {schema_to_ts(additional, schemas)}>"
+        return "Record<string, unknown>"
+
+    lines = ["{"]
+    for property_name in sorted(properties):
+        property_schema = properties[property_name]
+        optional = "" if property_name in required else "?"
+        lines.append(
+            f"{child_pad}{ts_property_name(property_name)}{optional}: "
+            f"{schema_to_ts(property_schema, schemas)};"
+        )
+    lines.append(f"{pad}}}")
+    return "\n".join(lines)
+
+
+def schema_to_ts(schema: dict[str, Any], schemas: dict[str, Any]) -> str:
+    if "$ref" in schema:
+        return schema["$ref"].removeprefix("#/components/schemas/")
+
+    if "enum" in schema:
+        values = schema["enum"]
+        return " | ".join(json.dumps(value) for value in values) or "never"
+
+    if "const" in schema:
+        return json.dumps(schema["const"])
+
+    if "anyOf" in schema:
+        return render_union(schema["anyOf"], schemas)
+
+    if "oneOf" in schema:
+        return render_union(schema["oneOf"], schemas)
+
+    if "allOf" in schema:
+        return " & ".join(schema_to_ts(item, schemas) for item in schema["allOf"])
+
+    schema_type = schema.get("type")
+    if isinstance(schema_type, list):
+        return " | ".join(
+            schema_type_to_ts(item, schema, schemas) for item in schema_type
+        )
+
+    return schema_type_to_ts(schema_type, schema, schemas)
+
+
+def render_union(items: list[dict[str, Any]], schemas: dict[str, Any]) -> str:
+    rendered = [schema_to_ts(item, schemas) for item in items]
+    return " | ".join(rendered) or "unknown"
+
+
+def schema_type_to_ts(
+    schema_type: str | None,
+    schema: dict[str, Any],
+    schemas: dict[str, Any],
+) -> str:
+    if schema_type == "string":
+        return "string"
+    if schema_type in {"integer", "number"}:
+        return "number"
+    if schema_type == "boolean":
+        return "boolean"
+    if schema_type == "null":
+        return "null"
+    if schema_type == "array":
+        return f"{array_item_to_ts(schema.get('items', {}), schemas)}[]"
+    if schema_type == "object" or "properties" in schema:
+        return render_object(schema, schemas, indent=0)
+    return "unknown"
+
+
+def array_item_to_ts(schema: dict[str, Any], schemas: dict[str, Any]) -> str:
+    rendered = schema_to_ts(schema, schemas)
+    if " | " in rendered or " & " in rendered:
+        return f"({rendered})"
+    return rendered
+
+
+def ts_property_name(name: str) -> str:
+    if name.isidentifier() and not keyword.iskeyword(name):
+        return name
+    return json.dumps(name)
+
+
+if __name__ == "__main__":
+    main()