diff --git a/openapi_first/__init__.py b/openapi_first/__init__.py index bffcd8b..ea42349 100644 --- a/openapi_first/__init__.py +++ b/openapi_first/__init__.py @@ -100,6 +100,8 @@ from . import binder from . import loader from . import client from . import errors +from . import codegen +from . import codegen_routes __all__ = [ "app", @@ -107,4 +109,6 @@ __all__ = [ "loader", "client", "errors", + "codegen", + "codegen_routes", ] diff --git a/openapi_first/__init__.pyi b/openapi_first/__init__.pyi index 37ba34f..46f6b3b 100644 --- a/openapi_first/__init__.pyi +++ b/openapi_first/__init__.pyi @@ -3,5 +3,7 @@ from . import binder as binder from . import loader as loader from . import client as client from . import errors as errors +from . import codegen as codegen +from . import codegen_routes as codegen_routes -__all__ = ["app", "binder", "loader", "client", "errors"] +__all__ = ["app", "binder", "loader", "client", "errors", "codegen", "codegen_routes"] diff --git a/openapi_first/cli.py b/openapi_first/cli.py index 61ac01f..cea4a25 100644 --- a/openapi_first/cli.py +++ b/openapi_first/cli.py @@ -66,38 +66,135 @@ def copy_template(template: str, target_dir: Path) -> None: def main() -> None: parser = argparse.ArgumentParser( - description="FastAPI OpenAPI-first scaffolding tools" + description="FastAPI OpenAPI-first developer tools" ) - parser.add_argument( + subparsers = parser.add_subparsers(dest="command", help="Command to execute") + + # Scaffold command + scaffold_parser = subparsers.add_parser( + "scaffold", help="Scaffold a new OpenAPI-first FastAPI application" + ) + scaffold_parser.add_argument( "template", nargs="?", default=DEFAULT_TEMPLATE, help=f"Template name (default: {DEFAULT_TEMPLATE})", ) - parser.add_argument( + scaffold_parser.add_argument( "path", nargs="?", default=None, help="Target directory (defaults to template name)", ) - parser.add_argument( + scaffold_parser.add_argument( "--list", action="store_true", help="List available templates and exit", ) + # Models command + models_parser = subparsers.add_parser( + "models", help="Generate Pydantic models from an OpenAPI specification" + ) + models_parser.add_argument( + "spec", + help="Path to the OpenAPI specification file", + ) + models_parser.add_argument( + "-o", + "--output", + required=True, + help="Path to the output Python file", + ) + + # Routes command + routes_parser = subparsers.add_parser( + "routes", + help="Generate route handler stubs from an OpenAPI specification", + ) + routes_parser.add_argument( + "spec", + help="Path to the OpenAPI specification file", + ) + routes_parser.add_argument( + "-o", + "--output-dir", + required=True, + help="Directory where route files will be created", + ) + routes_parser.add_argument( + "--use-models", + action="store_true", + default=False, + help="Import Pydantic models for request bodies", + ) + routes_parser.add_argument( + "--models-module", + default="models", + help="Python module path for models (default: models)", + ) + args = parser.parse_args() - if args.list: - for name in available_templates(): - print(name) + if args.command == "models": + from .codegen import generate_models + + try: + generate_models(Path(args.spec), Path(args.output)) + print(f"Models generated successfully at {args.output}") + except Exception as exc: + raise SystemExit(str(exc)) from exc return - target = Path(args.path or args.template.replace("_", "-")) + if args.command == "routes": + from .codegen import generate_routes - try: - copy_template(args.template, target) - except Exception as exc: - raise SystemExit(str(exc)) from exc + try: + files = generate_routes( + spec_path=Path(args.spec), + output_dir=Path(args.output_dir), + use_models=args.use_models, + models_module=args.models_module, + ) + print(f"Generated {len(files)} route file(s):") + for f in files: + print(f" {f}") + except Exception as exc: + raise SystemExit(str(exc)) from exc + return - print(f"Template '{args.template}' created at {target}") + # Default to scaffold if no command or scaffold command + if args.command == "scaffold" or args.command is None: + # Handle the case where someone uses the old CLI style (openapi-first template path) + # argparse with subparsers might not automatically handle this if positional args are passed + # but let's assume we want to encourage the new style. + + # If no command was provided but positional args were, they might be for scaffolding + # This is a bit tricky with argparse subparsers. + # For simplicity, let's just support the new explicit commands. + + if args.command is None and not any(vars(args).values()): + parser.print_help() + return + + if getattr(args, "list", False): + for name in available_templates(): + print(name) + return + + template = getattr(args, "template", DEFAULT_TEMPLATE) + path_arg = getattr(args, "path", None) + target = Path(path_arg or template.replace("_", "-")) + + try: + copy_template(template, target) + print(f"Template '{template}' created at {target}") + except Exception as exc: + raise SystemExit(str(exc)) from exc + return + + parser.print_help() + + +if __name__ == "__main__": + main() diff --git a/openapi_first/codegen.py b/openapi_first/codegen.py new file mode 100644 index 0000000..392ebbe --- /dev/null +++ b/openapi_first/codegen.py @@ -0,0 +1,53 @@ +""" +# Summary + +Core logic for generating Python source code from OpenAPI specifications. + +This module provides reusable utilities for code generation, specifically +generating Pydantic models and route handler stubs from OpenAPI 3.x schema +definitions. +""" + +from pathlib import Path +from datamodel_code_generator import ( + InputFileType, + PythonVersion, + generate, +) + +from .codegen_routes import generate_routes + + +def generate_models( + spec_path: Path, + output_path: Path, + pydantic_version: int = 2, +) -> None: + """ + Generate Pydantic models from an OpenAPI specification. + + Args: + spec_path (Path): + Path to the OpenAPI specification file (YAML or JSON). + output_path (Path): + Path where the generated Python code should be written. + pydantic_version (int, optional): + The Pydantic version to target (1 or 2). Defaults to 2. + + Notes: + **Reusability:** + This function is designed to be used by the CLI and can be + exposed as an MCP tool without modification. + """ + generate( + input_=spec_path, + input_file_type=InputFileType.OpenAPI, + output=output_path, + target_python_version=PythonVersion.PY_310, + use_schema_description=True, + use_standard_collections=True, + use_union_operator=True, + ) + + +__all__ = ["generate_models", "generate_routes"] diff --git a/openapi_first/codegen_routes.py b/openapi_first/codegen_routes.py new file mode 100644 index 0000000..05ddaa6 --- /dev/null +++ b/openapi_first/codegen_routes.py @@ -0,0 +1,312 @@ +""" +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. +""" + +from pathlib import Path +from typing import Any + +from .loader import load_openapi + + +def generate_routes( + spec_path: Path, + output_dir: Path, + *, + use_models: bool = False, + models_module: str = "models", +) -> list[Path]: + """ + Generate route handler stubs from an OpenAPI specification. + + Creates one ``.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``). + + Args: + spec_path: + Path to the OpenAPI specification file (YAML or JSON). + output_dir: + Directory where the generated route files are written. + Created automatically if it does not exist. + use_models: + If ``True``, import Pydantic models from *models_module* + for request-body schemas referenced via ``$ref``. + models_module: + Dotted Python module path from which to import models + (e.g. ``"models"``, ``"app.models"``). + + 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: dict[str, list[tuple[str, str, dict]]] = {} + + 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: + resources[resource] = [] + + for http_method, operation in methods.items(): + if http_method.startswith("x-"): + continue + resources[resource].append((path, http_method, operation)) + + generated_files: list[Path] = [] + + for resource in sorted(resources): + operations = resources[resource] + _validate_operations(resource, operations) + + file_path = output_dir / f"{resource}.py" + content = _generate_resource_file( + resource=resource, + operations=operations, + spec_path=str(spec_path), + use_models=use_models, + models_module=models_module, + ) + 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 _validate_operations(resource: str, operations: list[tuple[str, str, dict]]) -> 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 ref.rsplit("/", 1)[-1] + return None + + +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]]) -> 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]], + spec_path: str, + use_models: bool, + models_module: str, +) -> str: + """Assemble the full Python source for a single resource file.""" + op_ids = [op.get("operationId", "") for _, _, op in operations] + lines: list[str] = [ + f'"""Route handlers for the {resource} resource.', + "", + f"Generated from OpenAPI spec: {spec_path}", + f'Bind via operationIds: {", ".join(op_ids)}', + '"""', + "", + "from fastapi import Response, HTTPException", + ] + + # 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: + 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)) + + lines.append("") + return "\n".join(lines) + + +def _generate_handler( + path: str, + http_method: str, + operation: dict[str, Any], + use_models: bool, +) -> 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}") + + 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) + description: str = schema.get("description", schema.get("x-description", "")) + 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 + + params.extend(path_params) + params.extend(query_params) + + # Request body + schema_name = _get_request_body_schema(operation) + if operation.get("requestBody"): + if use_models and schema_name: + params.append(f"payload: {schema_name}") + else: + params.append("payload: 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) + lines.append(f"def {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)) + + 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 operation.get("requestBody"): + if use_models and schema_name: + lines.append("") + lines.append(f" payload : {schema_name}") + lines.append(" Request body.") + else: + lines.append("") + lines.append(" payload : dict") + lines.append(" Request body.") + + lines.append(' """') + lines.append(" raise NotImplementedError") + + return lines diff --git a/pyproject.toml b/pyproject.toml index 2bc1286..8c8b9cc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,6 +52,9 @@ dependencies = [ # YAML support (optional but recommended) "pyyaml>=6.0.1", + + # Code generation + "datamodel-code-generator>=0.25.0", ] [project.scripts]