feat: generate OpenAPI spec from event handler

Author: rubenfonseca

Diff for: aws_lambda_powertools/event_handler/api_gateway.py

@@ -4,7 +4,6 @@
 import re
 import traceback
 import warnings
-import zlib
 from abc import ABC, abstractmethod
 from enum import Enum
 from functools import partial
@@ -17,14 +16,23 @@
     Match,
     Optional,
     Pattern,
+    Sequence,
     Set,
     Tuple,
     Type,
     Union,
 )
 
+import zlib
+from pydantic.fields import ModelField
+from pydantic.schema import get_flat_models_from_fields, get_model_name_map, model_process_schema
+
 from aws_lambda_powertools.event_handler import content_types
 from aws_lambda_powertools.event_handler.exceptions import NotFoundError, ServiceError
+from aws_lambda_powertools.event_handler.openapi.dependant import get_dependant
+from aws_lambda_powertools.event_handler.openapi.models import Contact, License, OpenAPI, Server, Tag
+from aws_lambda_powertools.event_handler.openapi.utils import get_flat_params
+from aws_lambda_powertools.event_handler.route import Route
 from aws_lambda_powertools.shared.cookies import Cookie
 from aws_lambda_powertools.shared.functions import powertools_dev_is_set
 from aws_lambda_powertools.shared.json_encoder import Encoder
@@ -207,26 +215,6 @@ def __init__(
             self.headers.setdefault("Content-Type", content_type)
 
 
-class Route:
-    """Internally used Route Configuration"""
-
-    def __init__(
-        self,
-        method: str,
-        rule: Pattern,
-        func: Callable,
-        cors: bool,
-        compress: bool,
-        cache_control: Optional[str],
-    ):
-        self.method = method.upper()
-        self.rule = rule
-        self.func = func
-        self.cors = cors
-        self.compress = compress
-        self.cache_control = cache_control
-
-
 class ResponseBuilder:
     """Internally used Response builder"""
 
@@ -554,6 +542,119 @@ def __init__(
         # Allow for a custom serializer or a concise json serialization
         self._serializer = serializer or partial(json.dumps, separators=(",", ":"), cls=Encoder)
 
+    def get_openapi_schema(
+        self,
+        *,
+        title: str,
+        version: str,
+        openapi_version: str = "3.1.0",
+        summary: Optional[str] = None,
+        description: Optional[str] = None,
+        tags: Optional[List[Tag]] = None,
+        servers: Optional[List[Server]] = None,
+        terms_of_service: Optional[str] = None,
+        contact: Optional[Contact] = None,
+        license_info: Optional[License] = None,
+    ) -> OpenAPI:
+        info: Dict[str, Any] = {"title": title, "version": version}
+        if summary:
+            info["summary"] = summary
+        if description:
+            info["description"] = description
+        if terms_of_service:
+            info["termsOfService"] = terms_of_service
+        if contact:
+            info["contact"] = contact
+        if license_info:
+            info["license"] = license_info
+
+        output: Dict[str, Any] = {"openapi": openapi_version, "info": info}
+        if servers:
+            output["servers"] = servers
+        else:
+            # If the servers property is not provided, or is an empty array, the default value would be a Server Object
+            # with a url value of /.
+            output["servers"] = [Server(url="/")]
+
+        components: Dict[str, Dict[str, Any]] = {}
+        paths: Dict[str, Dict[str, Any]] = {}
+        operation_ids: Set[str] = set()
+
+        all_routes = self._dynamic_routes + self._static_routes
+        all_fields = self._get_fields_from_routes(all_routes)
+        models = get_flat_models_from_fields(all_fields, known_models=set())
+        model_name_map = get_model_name_map(models)
+
+        definitions: Dict[str, Dict[str, Any]] = {}
+        for model in models:
+            m_schema, m_definitions, _ = model_process_schema(
+                model,
+                model_name_map=model_name_map,
+                ref_prefix="#/components/schemas/",
+            )
+            definitions.update(m_definitions)
+            model_name = model_name_map[model]
+            if "description" in m_schema:
+                m_schema["description"] = m_schema["description"].split("\f")[0]
+            definitions[model_name] = m_schema
+
+        for route in all_routes:
+            dependant = get_dependant(
+                path=route.func.__name__,
+                call=route.func,
+            )
+
+            result = route._openapi_path(
+                dependant=dependant,
+                operation_ids=operation_ids,
+                model_name_map=model_name_map,
+            )
+            if result:
+                path, path_definitions = result
+                if path:
+                    paths.setdefault(route.path, {}).update(path)
+                if path_definitions:
+                    definitions.update(path_definitions)
+
+        if definitions:
+            components["schemas"] = {k: definitions[k] for k in sorted(definitions)}
+        if components:
+            output["components"] = components
+        if tags:
+            output["tags"] = tags
+
+        output["paths"] = paths
+
+        return OpenAPI(**output)  # .dict(by_alias=True, exclude_none=True)
+
+    def get_openapi_json_schema(
+        self,
+        *,
+        title: str,
+        version: str,
+        openapi_version: str = "3.1.0",
+        summary: Optional[str] = None,
+        description: Optional[str] = None,
+        tags: Optional[List[Tag]] = None,
+        servers: Optional[List[Server]] = None,
+        terms_of_service: Optional[str] = None,
+        contact: Optional[Contact] = None,
+        license_info: Optional[License] = None,
+    ) -> str:
+        """Returns the OpenAPI schema as a JSON serializable dict"""
+        return self.get_openapi_schema(
+            title=title,
+            version=version,
+            openapi_version=openapi_version,
+            summary=summary,
+            description=description,
+            tags=tags,
+            servers=servers,
+            terms_of_service=terms_of_service,
+            contact=contact,
+            license_info=license_info,
+        ).json(by_alias=True, exclude_none=True, indent=2)
+
     def route(
         self,
         rule: str,
@@ -573,7 +674,7 @@ def register_resolver(func: Callable):
                 cors_enabled = cors
 
             for item in methods:
-                _route = Route(item, self._compile_regex(rule), func, cors_enabled, compress, cache_control)
+                _route = Route(item, rule, self._compile_regex(rule), func, cors_enabled, compress, cache_control)
 
                 # The more specific route wins.
                 # We store dynamic (/studies/{studyid}) and static routes (/studies/fetch) separately.
@@ -889,6 +990,22 @@ def include_router(self, router: "Router", prefix: Optional[str] = None) -> None
 
             self.route(*new_route)(func)
 
+    @staticmethod
+    def _get_fields_from_routes(routes: Sequence[Route]) -> List[ModelField]:
+        responses_from_routes: List[ModelField] = []
+        request_fields_from_routes: List[ModelField] = []
+
+        for route in routes:
+            dependant = get_dependant(path=route.path, call=route.func)
+            params = get_flat_params(dependant)
+            request_fields_from_routes.extend(params)
+
+            if dependant.return_param:
+                responses_from_routes.append(dependant.return_param)
+
+        flat_models = list(responses_from_routes + request_fields_from_routes)
+        return flat_models
+
 
 class Router(BaseRouter):
     """Router helper class to allow splitting ApiGatewayResolver into multiple files"""

Diff for: aws_lambda_powertools/event_handler/openapi/__init__.py

@@ -0,0 +1,11 @@
+from aws_lambda_powertools.event_handler.openapi.models import (
+    Example,
+    Info,
+    MediaType,
+    Operation,
+    Reference,
+    Response,
+    Schema,
+)
+
+__all__ = ["Info", "Operation", "Response", "MediaType", "Reference", "Schema", "Example"]

Diff for: aws_lambda_powertools/event_handler/openapi/dependant.py

@@ -0,0 +1,100 @@
+import inspect
+import re
+from typing import Any, Callable, Dict, ForwardRef, Optional, Set, cast
+
+from pydantic.fields import ModelField
+from pydantic.typing import evaluate_forwardref
+
+from aws_lambda_powertools.event_handler.openapi.params import Dependant, Param, ParamTypes, analyze_param
+
+
+def add_param_to_fields(
+    *,
+    field: ModelField,
+    dependant: Dependant,
+) -> None:
+    field_info = cast(Param, field.field_info)
+    if field_info.in_ == ParamTypes.path:
+        dependant.path_params.append(field)
+    elif field_info.in_ == ParamTypes.query:
+        dependant.query_params.append(field)
+    elif field_info.in_ == ParamTypes.header:
+        dependant.header_params.append(field)
+    else:
+        assert field_info.in_ == ParamTypes.cookie
+        dependant.cookie_params.append(field)
+
+
+def get_typed_annotation(annotation: Any, globalns: Dict[str, Any]) -> Any:
+    if isinstance(annotation, str):
+        annotation = ForwardRef(annotation)
+        annotation = evaluate_forwardref(annotation, globalns, globalns)
+    return annotation
+
+
+def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
+    signature = inspect.signature(call)
+    globalns = getattr(call, "__global__", {})
+    typed_params = [
+        inspect.Parameter(
+            name=param.name,
+            kind=param.kind,
+            default=param.default,
+            annotation=get_typed_annotation(param.annotation, globalns),
+        )
+        for param in signature.parameters.values()
+    ]
+
+    if signature.return_annotation is not inspect.Signature.empty:
+        return_param = inspect.Parameter(
+            name="Return",
+            kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
+            default=None,
+            annotation=get_typed_annotation(signature.return_annotation, globalns),
+        )
+        return inspect.Signature(typed_params, return_annotation=return_param.annotation)
+    else:
+        return inspect.Signature(typed_params)
+
+
+def get_path_param_names(path: str) -> Set[str]:
+    return set(re.findall("{(.*?)}", path))
+
+
+def get_dependant(
+    *,
+    path: str,
+    call: Callable[..., Any],
+    name: Optional[str] = None,
+) -> Dependant:
+    path_param_names = get_path_param_names(path)
+    endpoint_signature = get_typed_signature(call)
+    signature_params = endpoint_signature.parameters
+    dependant = Dependant(
+        call=call,
+        name=name,
+        path=path,
+    )
+
+    for param_name, param in signature_params.items():
+        is_path_param = param_name in path_param_names
+        type_annotation, param_field = analyze_param(
+            param_name=param_name,
+            annotation=param.annotation,
+            value=param.default,
+            is_path_param=is_path_param,
+        )
+        assert param_field is not None
+
+        add_param_to_fields(field=param_field, dependant=dependant)
+
+    return_annotation = endpoint_signature.return_annotation
+    if return_annotation is not inspect.Signature.empty:
+        type_annotation, param_field = analyze_param(
+            param_name="Return", annotation=return_annotation, value=None, is_path_param=False,
+        )
+        assert param_field is not None
+
+        dependant.return_param = param_field
+
+    return dependant