aetos/openapi-first
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: codegen routes — sanitize resource names, --inject-request/--package, expand form bodies

Browse Source 
codegen_routes.generate_routes now sanitizes resource identifiers (fetch-requests -> fetch_requests), optionally prepends 'request: Request' (inject_request) and emits <resource>/routes.py packages with __init__.py (package). multipart/form-data and x-www-form-urlencoded bodies expand to per-property UploadFile/File/Form params instead of an opaque dict; spec paths are normalized to forward slashes to avoid Windows escape issues. Adds CLI flags --inject-request/--package, 7 focused tests, and refreshes lib/wiki/MCP docs to the real CLI.
This commit is contained in:
Vishesh 'ironeagle' Bangotra
2026-10-13 03:26:31 +05:30
parent b559323dfe
commit 51ffb4aae9
38 changed files with 1156 additions and 3027 deletions
Download Patch File Download Diff File Expand all files Collapse all files

182
openapi_first/codegen_routes.py
View File

@@ -18,13 +18,45 @@ Notes:
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``.
"""
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,
@@ -32,13 +64,15 @@ def generate_routes(
*,
use_models: bool = False,
models_module: str = "models",
inject_request: bool = False,
package: bool = False,
) -> list[Path]:
"""
Generate route handler stubs from an OpenAPI specification.
Creates one ``<resource>.py`` file per resource in *output_dir*.
Resources are derived from the first path segment (e.g. ``/pets``
and ``/pets/{id}`` both group under ``pets``).
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):
@@ -52,6 +86,12 @@ def generate_routes(
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.
Returns:
list[Path]:
@@ -86,17 +126,29 @@ def generate_routes(
generated_files: list[Path] = []
if package:
_ensure_package(output_dir)
for resource in sorted(resources_map):
operations = resources_map[resource]
_validate_operations(resource, operations)
file_path = output_dir / f"{resource}.py"
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,
)
file_path.write_text(content, encoding="utf-8")
generated_files.append(file_path)
@@ -118,6 +170,13 @@ _TYPE_MAP: dict[str, str] = {
}
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:
@@ -152,6 +211,55 @@ def _get_request_body_schema(operation: dict[str, Any]) -> str | None:
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", {})
@@ -184,9 +292,39 @@ def _generate_resource_file(
spec_path: str,
use_models: bool,
models_module: str,
inject_request: 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.',
"",
@@ -194,7 +332,7 @@ def _generate_resource_file(
f'Bind via operationIds: {", ".join(op_ids)}',
'"""',
"",
"from fastapi import Response, HTTPException",
f"from fastapi import {', '.join(fastapi_names)}",
]
# Conditional typing import
@@ -205,6 +343,8 @@ def _generate_resource_file(
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)
@@ -217,7 +357,11 @@ def _generate_resource_file(
for path, http_method, operation in operations:
lines.append("")
lines.extend(_generate_handler(path, http_method, operation, use_models))
lines.extend(
_generate_handler(
path, http_method, operation, use_models, inject_request
)
)
lines.append("")
return "\n".join(lines)
@@ -228,6 +372,7 @@ def _generate_handler(
http_method: str,
operation: dict[str, Any],
use_models: bool,
inject_request: bool = False,
) -> list[str]:
"""Build the source lines for a single handler function."""
operation_id = operation.get("operationId", "")
@@ -257,12 +402,27 @@ def _generate_handler(
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)
if operation.get("requestBody"):
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"payload: {schema_name}")
else:
@@ -303,7 +463,13 @@ def _generate_handler(
else:
lines.append(f" {pname} : {ptype}")
if operation.get("requestBody"):
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" payload : {schema_name}")