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

chore(release): 0.0.7

Browse Source 
Squash-merge lib-standardisation: adopt repo standards across packaging,
tooling, docs, and CI.

- standardize pyproject.toml (canonical packaging, lint config, extras, GSDFC pydoclint)
- add py.typed marker, modernize type hints and Google-style docstrings
- nest generated docs under docs/lib and docs/mcp; add hand-written wiki
- adopt standard Type A PyPI CI pipeline (tag==version guard + quality gate)
- promote generated CHANGELOG.md; bump version to 0.0.7
This commit is contained in:
Vishesh 'ironeagle' Bangotra
2026-10-17 02:16:49 +05:30
parent c95de5a6e9
commit 621f10409e
162 changed files with 10165 additions and 2846 deletions
Download Patch File Download Diff File Expand all files Collapse all files

272
openapi_first/codegen_routes.py
View File

@@ -1,4 +1,6 @@
"""
# Summary
Route handler code generation from OpenAPI specifications.
This module generates Python route handler stubs from an OpenAPI 3.x
@@ -16,13 +18,48 @@ 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``.
- 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,
@@ -30,26 +67,40 @@ def generate_routes(
*,
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 ``<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:
spec_path (Path):
Path to the OpenAPI specification file (YAML or JSON).
output_dir:
output_dir (Path):
Directory where the generated route files are written.
Created automatically if it does not exist.
use_models:
use_models (bool, optional):
If ``True``, import Pydantic models from *models_module*
for request-body schemas referenced via ``$ref``.
models_module:
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]:
@@ -66,7 +117,7 @@ def generate_routes(
output_dir.mkdir(parents=True, exist_ok=True)
# Group paths by resource (first non-param path segment)
resources: dict[str, list[tuple[str, str, dict]]] = {}
resources_map: dict[str, list[tuple[str, str, dict[str, Any]]]] = {}
paths = spec.get("paths", {})
for path, methods in paths.items():
@@ -74,27 +125,40 @@ def generate_routes(
if not segments:
continue
resource = segments[0]
if resource not in resources:
resources[resource] = []
if resource not in resources_map:
resources_map[resource] = []
for http_method, operation in methods.items():
if http_method.startswith("x-"):
continue
resources[resource].append((path, http_method, operation))
resources_map[resource].append((path, http_method, operation))
generated_files: list[Path] = []
for resource in sorted(resources):
operations = resources[resource]
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,
template_style=template_style,
)
file_path.write_text(content, encoding="utf-8")
generated_files.append(file_path)
@@ -116,7 +180,16 @@ _TYPE_MAP: dict[str, str] = {
}
def _validate_operations(resource: str, operations: list[tuple[str, str, dict]]) -> None:
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"):
@@ -144,10 +217,59 @@ def _get_request_body_schema(operation: dict[str, Any]) -> str | None:
schema = media_info.get("schema", {})
ref = schema.get("$ref", "")
if ref:
return ref.rsplit("/", 1)[-1]
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", {})
@@ -159,7 +281,7 @@ def _get_success_status(operation: dict[str, Any]) -> str:
return "200"
def _needs_any(operations: list[tuple[str, str, dict]]) -> bool:
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", []):
@@ -176,13 +298,44 @@ def _needs_any(operations: list[tuple[str, str, dict]]) -> bool:
def _generate_resource_file(
resource: str,
operations: list[tuple[str, str, dict]],
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.',
"",
@@ -190,7 +343,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
@@ -201,17 +354,31 @@ 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)
if schemas_needed:
lines.append(f"from {models_module} import {', '.join(sorted(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))
lines.extend(
_generate_handler(
path,
http_method,
operation,
use_models,
inject_request,
resource,
template_style,
)
)
lines.append("")
return "\n".join(lines)
@@ -222,10 +389,14 @@ def _generate_handler(
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] = []
@@ -237,7 +408,6 @@ def _generate_handler(
schema: dict[str, Any] = param.get("schema", {})
param_type: str = _resolve_type(schema)
required: bool = param.get("required", False)
description: str = schema.get("description", schema.get("x-description", ""))
default_raw = schema.get("default")
if param_in == "path":
@@ -252,16 +422,31 @@ 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}")
params.append(f"{body_param}: {schema_name}")
else:
params.append("payload: dict")
params.append(f"{body_param}: dict")
# Inject Response for non-200 success codes
success_status = _get_success_status(operation)
@@ -271,19 +456,22 @@ def _generate_handler(
# Build function body
lines: list[str] = []
param_str = ", ".join(params)
lines.append(f"def {operation_id}({param_str}):")
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", []):
name: str = param.get("name", "")
param_in: str = param.get("in", "")
schema: dict[str, Any] = param.get("schema", {})
param_type: str = _resolve_type(schema)
description: str = param.get("description", schema.get("x-description", ""))
if param_in in ("path", "query"):
doc_params.append((name, param_type, description))
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("")
@@ -296,17 +484,29 @@ 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}")
lines.append(f" {body_param} : {schema_name}")
lines.append(" Request body.")
else:
lines.append("")
lines.append(" payload : dict")
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