#!/usr/bin/env python3 """Generate OpenAPI JSON from Drogon routes plus a manual OpenAPI YAML file.""" from __future__ import annotations import argparse import copy import json import re from pathlib import Path from typing import Any ROUTE_RE = re.compile( r"ADD_METHOD_TO\(\s*([A-Za-z_]\w*)::([A-Za-z_]\w*)\s*,\s*" r'"([^"]+)"\s*,\s*drogon::([A-Za-z]+)\s*\)', re.MULTILINE | re.DOTALL, ) METHODS = {"get", "post", "put", "patch", "delete"} def controller_tag(controller: str) -> str: return controller.removesuffix("Controller") def operation_id(controller: str, handler: str) -> str: return f"{controller}.{handler}" def openapi_path(path: str) -> tuple[str, list[dict[str, Any]]]: parameters: list[dict[str, Any]] = [] segments = path.split("/") placeholder_names: dict[int, str] = {} for position, segment in enumerate(segments): match = re.fullmatch(r"\{(\d+)\}", segment) if not match: continue index = int(match.group(1)) if index == 1: placeholder_names[index] = "id" continue previous = segments[position - 1] if position > 0 else "" if previous.endswith("ies"): stem = previous[:-3] + "y" elif previous.endswith("s") and len(previous) > 1: stem = previous[:-1] else: stem = previous or f"param{index}" placeholder_names[index] = f"{stem}_id" def replace(match: re.Match[str]) -> str: index = int(match.group(1)) name = placeholder_names.get(index, f"param{index}") parameters.append( { "name": name, "in": "path", "required": True, "schema": {"type": "integer", "format": "int64"}, } ) return "{" + name + "}" return re.sub(r"\{(\d+)\}", replace, path), parameters def generated_operation( controller: str, handler: str, parameters: list[dict[str, Any]], ) -> dict[str, Any]: return { "tags": [controller_tag(controller)], "operationId": operation_id(controller, handler), "summary": handler_to_summary(handler), "parameters": parameters, "responses": {"200": {"description": "Response"}}, } def handler_to_summary(handler: str) -> str: words = re.sub(r"([a-z0-9])([A-Z])", r"\1 \2", handler).replace("_", " ") return words[:1].upper() + words[1:] def parse_routes(controllers_dir: Path) -> dict[tuple[str, str], dict[str, Any]]: operations: dict[tuple[str, str], dict[str, Any]] = {} for header in sorted(controllers_dir.glob("*.h")): text = header.read_text(encoding="utf-8") for controller, handler, raw_path, drogon_method in ROUTE_RE.findall(text): method = drogon_method.lower() path, parameters = openapi_path(raw_path) key = (path, method) if key in operations: raise ValueError(f"duplicate route for {method.upper()} {path}") operations[key] = generated_operation(controller, handler, parameters) return operations def load_manual(path: Path | None) -> dict[str, Any]: if path is None: return {} text = path.read_text(encoding="utf-8") try: import yaml # type: ignore[import-untyped] except ImportError as error: try: data = json.loads(text) except json.JSONDecodeError: raise SystemExit( "PyYAML is required unless the manual OpenAPI YAML uses " "JSON-compatible syntax" ) from error else: data = yaml.safe_load(text) if data is None: return {} if not isinstance(data, dict): raise ValueError(f"{path}: OpenAPI document must be an object") return data def merge_openapi( generated: dict[tuple[str, str], dict[str, Any]], manual: dict[str, Any], title: str, version: str, server: str, ) -> dict[str, Any]: spec: dict[str, Any] = { "openapi": manual.get("openapi", "3.0.3"), "info": copy.deepcopy( manual.get("info", {"title": title, "version": version}) ), "servers": copy.deepcopy(manual.get("servers", [{"url": server}])), "paths": {}, } for key in ("components", "tags", "externalDocs"): if key in manual: spec[key] = copy.deepcopy(manual[key]) manual_paths = manual.get("paths", {}) if not isinstance(manual_paths, dict): raise ValueError("manual OpenAPI paths must be an object") seen_manual: set[tuple[str, str]] = set() for path, path_item in manual_paths.items(): if not isinstance(path_item, dict): raise ValueError(f"{path}: path item must be an object") for method, manual_operation in path_item.items(): method_lower = method.lower() if method_lower not in METHODS: continue key = (path, method_lower) seen_manual.add(key) if key not in generated and not manual_operation.get("x-manual-only", False): raise ValueError( f"{method_upper(method_lower)} {path}: manual operation has no matching Drogon route" ) for key, operation in sorted(generated.items()): path, method = key manual_operation = manual_paths.get(path, {}).get(method, {}) merged = copy.deepcopy(operation) merged.update(copy.deepcopy(manual_operation)) merged.pop("x-manual-only", None) spec["paths"].setdefault(path, {})[method] = merged for key in sorted(seen_manual - generated.keys()): path, method = key manual_operation = copy.deepcopy(manual_paths[path][method]) if not manual_operation.get("x-manual-only", False): continue manual_operation.pop("x-manual-only", None) spec["paths"].setdefault(path, {})[method] = manual_operation return spec def method_upper(method: str) -> str: return method.upper() def generate( controllers_dir: Path, manual_path: Path | None, title: str, version: str, server: str, ) -> dict[str, Any]: generated = parse_routes(controllers_dir) manual = load_manual(manual_path) return merge_openapi(generated, manual, title, version, server) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--controllers", type=Path, required=True) parser.add_argument("--manual", type=Path) parser.add_argument("--out", type=Path, required=True) parser.add_argument("--title", default="Visione API") parser.add_argument("--version", default="0.1.0") parser.add_argument("--server", default="http://localhost:5555") args = parser.parse_args() spec = generate( controllers_dir=args.controllers, manual_path=args.manual, title=args.title, version=args.version, server=args.server, ) args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text( json.dumps(spec, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) if __name__ == "__main__": main()