Vishesh 'ironeagle' Bangotra
9cb8bf8f1c
Add generate_routes(template_style=...) and CLI --template-style. When on, stubs use async def, a JSON body parameter named body (dict or Model), and with --inject-request a service-lookup hint comment before NotImplementedError. Default behaviour (def/payload) unchanged. +2 tests (32 passed), wiki + MCP bundle regenerated.
508 lines
18 KiB
Python
508 lines
18 KiB
Python
"""
|
|
# Summary
|
|
|
|
Route handler code generation from OpenAPI specifications.
|
|
|
|
This module generates Python route handler stubs from an OpenAPI 3.x
|
|
specification. Each resource (derived from the first path segment)
|
|
gets its own file under the output directory. Every OpenAPI operation
|
|
must define an ``operationId``, which becomes the handler function name.
|
|
|
|
Notes:
|
|
**Design constraints:**
|
|
|
|
- ``operationId`` is required on every operation (matching
|
|
``binder.bind_routes``).
|
|
- Handlers are stubs raising ``NotImplementedError``.
|
|
- Sub-resources (e.g. ``/pets/{id}/photo``) are grouped with their
|
|
parent resource (``pets``).
|
|
- Parameter types and defaults are inferred from the spec.
|
|
- ``response: Response`` is injected for non-200 success codes.
|
|
- Resource names are sanitized to valid Python module identifiers,
|
|
so a path such as ``/fetch-requests`` yields ``fetch_requests.py``.
|
|
- Passing ``inject_request=True`` prepends ``request: Request`` to
|
|
every handler (the composition-root / DI convention).
|
|
- ``multipart/form-data`` and
|
|
``application/x-www-form-urlencoded`` request bodies expand to one
|
|
parameter per schema property (``UploadFile`` / ``File`` / ``Form``)
|
|
instead of an opaque ``payload: dict``.
|
|
- Passing ``package=True`` writes ``<resource>/routes.py`` (with
|
|
``__init__.py`` sidecars) instead of a flat ``<resource>.py``.
|
|
- Passing ``template_style=True`` emits ``async def`` handlers whose
|
|
JSON body parameter is named ``body`` (matching the standards
|
|
template) and adds a service-resolution hint comment.
|
|
"""
|
|
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .loader import load_openapi
|
|
|
|
_FORM_URLENCODED = "application/x-www-form-urlencoded"
|
|
_MULTIPART = "multipart/form-data"
|
|
|
|
|
|
def _sanitize_identifier(name: str) -> str:
|
|
"""Convert an arbitrary path segment into a valid Python module name.
|
|
|
|
Args:
|
|
name (str): Raw resource name (first path segment).
|
|
|
|
Returns:
|
|
str: A valid lower-case Python identifier (e.g. ``fetch-requests``
|
|
becomes ``fetch_requests``).
|
|
"""
|
|
ident = re.sub(r"\W", "_", name).strip("_").lower()
|
|
if not ident:
|
|
return "resource"
|
|
if ident[0].isdigit():
|
|
ident = f"_{ident}"
|
|
return ident
|
|
|
|
|
|
def generate_routes(
|
|
spec_path: Path,
|
|
output_dir: Path,
|
|
*,
|
|
use_models: bool = False,
|
|
models_module: str = "models",
|
|
inject_request: bool = False,
|
|
package: bool = False,
|
|
template_style: bool = False,
|
|
) -> list[Path]:
|
|
"""
|
|
Generate route handler stubs from an OpenAPI specification.
|
|
|
|
Creates one route module per resource in *output_dir*. Resources are
|
|
derived from the first path segment (e.g. ``/pets`` and ``/pets/{id}``
|
|
both group under ``pets``) and sanitized to valid module identifiers.
|
|
|
|
Args:
|
|
spec_path (Path):
|
|
Path to the OpenAPI specification file (YAML or JSON).
|
|
output_dir (Path):
|
|
Directory where the generated route files are written.
|
|
Created automatically if it does not exist.
|
|
use_models (bool, optional):
|
|
If ``True``, import Pydantic models from *models_module*
|
|
for request-body schemas referenced via ``$ref``.
|
|
models_module (str, optional):
|
|
Dotted Python module path from which to import models
|
|
(e.g. ``"models"``, ``"app.models"``).
|
|
inject_request (bool, optional):
|
|
If ``True``, prepend ``request: Request`` to every handler so the
|
|
stub can later resolve services from ``request.app.state``.
|
|
package (bool, optional):
|
|
If ``True``, emit ``<resource>/routes.py`` packages (each with an
|
|
``__init__.py``) instead of flat ``<resource>.py`` modules.
|
|
template_style (bool, optional):
|
|
If ``True``, emit ``async def`` handlers with a JSON body parameter
|
|
named ``body`` (the standards-template convention) plus a
|
|
service-resolution hint comment, instead of sync ``def`` handlers
|
|
taking ``payload``.
|
|
|
|
Returns:
|
|
list[Path]:
|
|
Absolute paths of every generated route file.
|
|
|
|
Raises:
|
|
OpenAPISpecLoadError:
|
|
If the spec cannot be loaded or validated.
|
|
ValueError:
|
|
If any operation is missing ``operationId``.
|
|
"""
|
|
spec = load_openapi(spec_path)
|
|
output_dir = Path(output_dir).resolve()
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Group paths by resource (first non-param path segment)
|
|
resources_map: dict[str, list[tuple[str, str, dict[str, Any]]]] = {}
|
|
|
|
paths = spec.get("paths", {})
|
|
for path, methods in paths.items():
|
|
segments = [s for s in path.split("/") if s and not s.startswith("{")]
|
|
if not segments:
|
|
continue
|
|
resource = segments[0]
|
|
if resource not in resources_map:
|
|
resources_map[resource] = []
|
|
|
|
for http_method, operation in methods.items():
|
|
if http_method.startswith("x-"):
|
|
continue
|
|
resources_map[resource].append((path, http_method, operation))
|
|
|
|
generated_files: list[Path] = []
|
|
|
|
if package:
|
|
_ensure_package(output_dir)
|
|
|
|
for resource in sorted(resources_map):
|
|
operations = resources_map[resource]
|
|
_validate_operations(resource, operations)
|
|
|
|
identifier = _sanitize_identifier(resource)
|
|
if package:
|
|
resource_dir = output_dir / identifier
|
|
resource_dir.mkdir(parents=True, exist_ok=True)
|
|
_ensure_package(resource_dir)
|
|
file_path = resource_dir / "routes.py"
|
|
else:
|
|
file_path = output_dir / f"{identifier}.py"
|
|
|
|
content = _generate_resource_file(
|
|
resource=resource,
|
|
operations=operations,
|
|
spec_path=str(spec_path),
|
|
use_models=use_models,
|
|
models_module=models_module,
|
|
inject_request=inject_request,
|
|
template_style=template_style,
|
|
)
|
|
file_path.write_text(content, encoding="utf-8")
|
|
generated_files.append(file_path)
|
|
|
|
return generated_files
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_TYPE_MAP: dict[str, str] = {
|
|
"integer": "int",
|
|
"number": "float",
|
|
"boolean": "bool",
|
|
"string": "str",
|
|
"array": "list",
|
|
"object": "dict",
|
|
}
|
|
|
|
|
|
def _ensure_package(directory: Path) -> None:
|
|
"""Create an ``__init__.py`` in *directory* when it is missing."""
|
|
init = directory / "__init__.py"
|
|
if not init.exists():
|
|
init.write_text("", encoding="utf-8")
|
|
|
|
|
|
def _validate_operations(
|
|
resource: str, operations: list[tuple[str, str, dict[str, Any]]]
|
|
) -> None:
|
|
"""Ensure every operation has an operationId."""
|
|
for path, http_method, operation in operations:
|
|
if not operation.get("operationId"):
|
|
raise ValueError(
|
|
f"Missing operationId for {http_method.upper()} {path} "
|
|
f"in resource '{resource}'. "
|
|
"All operations must have an explicit operationId."
|
|
)
|
|
|
|
|
|
def _resolve_type(schema: dict[str, Any]) -> str:
|
|
"""Map an OpenAPI schema type to a Python type annotation."""
|
|
openapi_type = schema.get("type", "")
|
|
return _TYPE_MAP.get(openapi_type, "str")
|
|
|
|
|
|
def _get_request_body_schema(operation: dict[str, Any]) -> str | None:
|
|
"""Extract the ``$ref`` schema name from a request body, if any."""
|
|
request_body = operation.get("requestBody")
|
|
if not request_body:
|
|
return None
|
|
|
|
content = request_body.get("content", {})
|
|
for media_info in content.values():
|
|
schema = media_info.get("schema", {})
|
|
ref = schema.get("$ref", "")
|
|
if ref:
|
|
return str(ref.rsplit("/", 1)[-1])
|
|
return None
|
|
|
|
|
|
def _get_form_properties(
|
|
operation: dict[str, Any],
|
|
) -> tuple[str, dict[str, Any], list[str]] | None:
|
|
"""Return ``(media_type, properties, required)`` for a form-style body.
|
|
|
|
Handles ``multipart/form-data`` and ``application/x-www-form-urlencoded``
|
|
request bodies whose schema is an inline object with ``properties``.
|
|
Returns ``None`` for JSON bodies or bodies that use ``$ref``.
|
|
"""
|
|
request_body = operation.get("requestBody")
|
|
if not request_body:
|
|
return None
|
|
|
|
content = request_body.get("content", {})
|
|
for media_type, media_info in content.items():
|
|
base = media_type.split(";", 1)[0].strip()
|
|
if base not in (_MULTIPART, _FORM_URLENCODED):
|
|
continue
|
|
schema = media_info.get("schema", {})
|
|
if "$ref" in schema:
|
|
return None
|
|
properties = schema.get("properties")
|
|
if not isinstance(properties, dict) or not properties:
|
|
return None
|
|
return base, properties, list(schema.get("required", []))
|
|
return None
|
|
|
|
|
|
def _is_binary(schema: dict[str, Any]) -> bool:
|
|
"""Whether a schema represents an uploaded binary payload."""
|
|
if schema.get("format") == "binary":
|
|
return True
|
|
if schema.get("type") == "array":
|
|
return _is_binary(schema.get("items", {}))
|
|
return False
|
|
|
|
|
|
def _form_annotation(schema: dict[str, Any]) -> str:
|
|
"""Map a form field schema to its Python annotation."""
|
|
if schema.get("type") == "array" and _is_binary(schema.get("items", {})):
|
|
return "list[UploadFile]"
|
|
if _is_binary(schema):
|
|
return "UploadFile"
|
|
ref = schema.get("$ref", "")
|
|
if ref:
|
|
return str(ref.rsplit("/", 1)[-1])
|
|
return _resolve_type(schema)
|
|
|
|
|
|
def _get_success_status(operation: dict[str, Any]) -> str:
|
|
"""Return the primary success status code (200, 201, 204, or default)."""
|
|
responses = operation.get("responses", {})
|
|
for code in ("201", "200", "204"):
|
|
if code in responses:
|
|
return code
|
|
if "default" in responses:
|
|
return "default"
|
|
return "200"
|
|
|
|
|
|
def _needs_any(operations: list[tuple[str, str, dict[str, Any]]]) -> bool:
|
|
"""Check if any operation uses a type that requires `from typing import Any`."""
|
|
for _, _, op in operations:
|
|
for param in op.get("parameters", []):
|
|
schema = param.get("schema", {})
|
|
if schema.get("type", "") not in _TYPE_MAP:
|
|
return True
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# File / function generation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _generate_resource_file(
|
|
resource: str,
|
|
operations: list[tuple[str, str, dict[str, Any]]],
|
|
spec_path: str,
|
|
use_models: bool,
|
|
models_module: str,
|
|
inject_request: bool = False,
|
|
template_style: bool = False,
|
|
) -> str:
|
|
"""Assemble the full Python source for a single resource file."""
|
|
op_ids = [op.get("operationId", "") for _, _, op in operations]
|
|
# Windows paths would inject invalid escapes (e.g. ``\U``) into the
|
|
# generated docstring; normalize to forward slashes for the header.
|
|
spec_path = spec_path.replace("\\", "/")
|
|
|
|
need_upload = False
|
|
need_file_form = False
|
|
need_form = False
|
|
for _, _, op in operations:
|
|
form = _get_form_properties(op)
|
|
if not form:
|
|
continue
|
|
_, properties, _ = form
|
|
for prop_schema in properties.values():
|
|
if _is_binary(prop_schema):
|
|
need_upload = True
|
|
need_file_form = True
|
|
else:
|
|
need_form = True
|
|
|
|
fastapi_names = ["Response", "HTTPException"]
|
|
if inject_request:
|
|
fastapi_names.insert(0, "Request")
|
|
if need_upload:
|
|
fastapi_names.append("UploadFile")
|
|
if need_file_form:
|
|
fastapi_names.append("File")
|
|
if need_form:
|
|
fastapi_names.append("Form")
|
|
|
|
lines: list[str] = [
|
|
f'"""Route handlers for the {resource} resource.',
|
|
"",
|
|
f"Generated from OpenAPI spec: {spec_path}",
|
|
f'Bind via operationIds: {", ".join(op_ids)}',
|
|
'"""',
|
|
"",
|
|
f"from fastapi import {', '.join(fastapi_names)}",
|
|
]
|
|
|
|
# Conditional typing import
|
|
if _needs_any(operations):
|
|
lines.append("from typing import Any")
|
|
|
|
# Conditional model imports
|
|
if use_models:
|
|
schemas_needed: set[str] = set()
|
|
for _, _, op in operations:
|
|
if _get_form_properties(op):
|
|
continue
|
|
schema = _get_request_body_schema(op)
|
|
if schema:
|
|
schemas_needed.add(schema)
|
|
if schemas_needed:
|
|
lines.append(
|
|
f"from {models_module} import {', '.join(sorted(schemas_needed))}"
|
|
)
|
|
|
|
lines.append("")
|
|
|
|
for path, http_method, operation in operations:
|
|
lines.append("")
|
|
lines.extend(
|
|
_generate_handler(
|
|
path, http_method, operation, use_models, inject_request,
|
|
resource, template_style,
|
|
)
|
|
)
|
|
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _generate_handler(
|
|
path: str,
|
|
http_method: str,
|
|
operation: dict[str, Any],
|
|
use_models: bool,
|
|
inject_request: bool = False,
|
|
resource: str = "",
|
|
template_style: bool = False,
|
|
) -> list[str]:
|
|
"""Build the source lines for a single handler function."""
|
|
operation_id = operation.get("operationId", "")
|
|
summary = operation.get("summary", f"{http_method.upper()} {path}")
|
|
body_param = "body" if template_style else "payload"
|
|
|
|
params: list[str] = []
|
|
path_params: list[str] = []
|
|
query_params: list[str] = []
|
|
|
|
for param in operation.get("parameters", []):
|
|
name: str = param.get("name", "")
|
|
param_in: str = param.get("in", "")
|
|
schema: dict[str, Any] = param.get("schema", {})
|
|
param_type: str = _resolve_type(schema)
|
|
required: bool = param.get("required", False)
|
|
default_raw = schema.get("default")
|
|
|
|
if param_in == "path":
|
|
path_params.append(f"{name}: {param_type}")
|
|
elif param_in == "query":
|
|
if default_raw is not None:
|
|
default_repr = repr(default_raw)
|
|
query_params.append(f"{name}: {param_type} = {default_repr}")
|
|
elif required:
|
|
query_params.append(f"{name}: {param_type}")
|
|
else:
|
|
query_params.append(f"{name}: {param_type} = None")
|
|
# header / cookie params could be extended here
|
|
|
|
if inject_request:
|
|
params.append("request: Request")
|
|
params.extend(path_params)
|
|
params.extend(query_params)
|
|
|
|
# Request body
|
|
form = _get_form_properties(operation)
|
|
schema_name = _get_request_body_schema(operation)
|
|
is_form = False
|
|
if form:
|
|
is_form = True
|
|
_, properties, required_fields = form
|
|
for field_name, field_schema in properties.items():
|
|
annotation = _form_annotation(field_schema)
|
|
is_file = _is_binary(field_schema)
|
|
factory = "File" if is_file else "Form"
|
|
if field_name in required_fields:
|
|
params.append(f"{field_name}: {annotation} = {factory}(...)")
|
|
else:
|
|
params.append(f"{field_name}: {annotation} = {factory}(None)")
|
|
elif operation.get("requestBody"):
|
|
if use_models and schema_name:
|
|
params.append(f"{body_param}: {schema_name}")
|
|
else:
|
|
params.append(f"{body_param}: dict")
|
|
|
|
# Inject Response for non-200 success codes
|
|
success_status = _get_success_status(operation)
|
|
if success_status in ("201", "204"):
|
|
params.append("response: Response")
|
|
|
|
# Build function body
|
|
lines: list[str] = []
|
|
param_str = ", ".join(params)
|
|
prefix = "async def" if template_style else "def"
|
|
lines.append(f"{prefix} {operation_id}({param_str}):")
|
|
lines.append(f' """{summary}')
|
|
|
|
# Document parameters
|
|
doc_params: list[tuple[str, str, str]] = []
|
|
for param in operation.get("parameters", []):
|
|
doc_name: str = param.get("name", "")
|
|
doc_in: str = param.get("in", "")
|
|
doc_schema: dict[str, Any] = param.get("schema", {})
|
|
doc_type: str = _resolve_type(doc_schema)
|
|
doc_description: str = param.get(
|
|
"description", doc_schema.get("x-description", "")
|
|
)
|
|
if doc_in in ("path", "query"):
|
|
doc_params.append((doc_name, doc_type, doc_description))
|
|
|
|
if doc_params:
|
|
lines.append("")
|
|
lines.append(" Parameters")
|
|
lines.append(" ----------")
|
|
for pname, ptype, desc in doc_params:
|
|
if desc:
|
|
lines.append(f" {pname} : {ptype}")
|
|
lines.append(f" {desc}")
|
|
else:
|
|
lines.append(f" {pname} : {ptype}")
|
|
|
|
if is_form:
|
|
_, properties, _ = form # type: ignore[misc]
|
|
lines.append("")
|
|
for field_name, field_schema in properties.items():
|
|
lines.append(f" {field_name} : {_form_annotation(field_schema)}")
|
|
lines.append(" Multipart form field.")
|
|
elif operation.get("requestBody"):
|
|
if use_models and schema_name:
|
|
lines.append("")
|
|
lines.append(f" {body_param} : {schema_name}")
|
|
lines.append(" Request body.")
|
|
else:
|
|
lines.append("")
|
|
lines.append(f" {body_param} : dict")
|
|
lines.append(" Request body.")
|
|
|
|
lines.append(' """')
|
|
if template_style and inject_request:
|
|
svc_key = f"{_sanitize_identifier(resource)}_svc"
|
|
lines.append(
|
|
f' # Resolve the service: '
|
|
f'request.app.state.db_resources["{svc_key}"]'
|
|
)
|
|
lines.append(" raise NotImplementedError")
|
|
|
|
return lines
|