226 lines
7.0 KiB
Python
226 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
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()
|