fix: customize the handler methods

Author: rubenfonseca

aws_lambda_powertools/event_handler/api_gateway.py

@@ -50,6 +50,7 @@
 # API GW/ALB decode non-safe URI chars; we must support them too
 _UNSAFE_URI = r"%<> \[\]{}|^"
 _NAMED_GROUP_BOUNDARY_PATTERN = rf"(?P\1[{_SAFE_URI}{_UNSAFE_URI}\\w]+)"
+_DEFAULT_OPENAPI_RESPONSE_DESCRIPTION = "Successful Response"
 _ROUTE_REGEX = "^{}$"
 
 if TYPE_CHECKING:
@@ -201,9 +202,13 @@ def __init__(
         cors: bool,
         compress: bool,
         cache_control: Optional[str],
-        middlewares: Optional[List[Callable[..., Response]]],
+        summary: Optional[str],
         description: Optional[str],
+        responses: Optional[Dict[Union[int, str], Dict[str, Any]]],
+        response_description: Optional[str],
         tags: Optional[List["Tag"]],
+        operation_id: Optional[str],
+        middlewares: Optional[List[Callable[..., Response]]],
     ):
         """
 
@@ -212,10 +217,10 @@ def __init__(
 
         method: str
             The HTTP method, example "GET"
-        rule: Pattern
-            The route rule, example "/my/path"
         path: str
             The path of the route
+        rule: Pattern
+            The route rule, example "/my/path"
         func: Callable
             The route handler function
         cors: bool
@@ -224,12 +229,20 @@ def __init__(
             Whether or not to enable gzip compression for this route
         cache_control: Optional[str]
             The cache control header value, example "max-age=3600"
-        middlewares: Optional[List[Callable[..., Response]]]
-            The list of route middlewares to be called in order.
+        summary: Optional[str]
+            The OpenAPI summary for this route
         description: Optional[str]
             The OpenAPI description for this route
+        responses: Optional[Dict[Union[int, str], Dict[str, Any]]]
+            The OpenAPI responses for this route
+        response_description: Optional[str]
+            The OpenAPI response description for this route
         tags: Optional[List[Tag]]
             The list of OpenAPI tags to be used for this route
+        operation_id: Optional[str]
+            The OpenAPI operationId for this route
+        middlewares: Optional[List[Callable[..., Response]]]
+            The list of route middlewares to be called in order.
         """
         self.method = method.upper()
         self.path = path
@@ -239,10 +252,13 @@ def __init__(
         self.cors = cors
         self.compress = compress
         self.cache_control = cache_control
-        self.middlewares = middlewares or []
+        self.summary = summary
         self.description = description
+        self.responses = responses
+        self.response_description = response_description
         self.tags = tags or []
-        self.operation_id = self.method.title() + self.func.__name__.title()
+        self.middlewares = middlewares or []
+        self.operation_id = operation_id or (self.method.title() + self.func.__name__.title())
 
         # _middleware_stack_built is used to ensure the middleware stack is only built once.
         self._middleware_stack_built = False
@@ -370,7 +386,7 @@ def _get_openapi_path(
 
         responses = operation.setdefault("responses", {})
         success_response = responses.setdefault("200", {})
-        success_response["description"] = "Success"
+        success_response["description"] = self.response_description or _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION
         success_response["content"] = {"application/json": {"schema": {}}}
         json_response = success_response["content"].setdefault("application/json", {})
 
@@ -389,7 +405,7 @@ def _get_openapi_path(
         return path, definitions
 
     def _openapi_operation_summary(self) -> str:
-        return f"{self.method.upper()} {self.path}"
+        return self.summary or f"{self.method.upper()} {self.path}"
 
     def _openapi_operation_metadata(self, operation_ids: Set[str]) -> Dict[str, Any]:
         operation: Dict[str, Any] = {}
@@ -596,8 +612,12 @@ def route(
         cors: Optional[bool] = None,
         compress: bool = False,
         cache_control: Optional[str] = None,
+        summary: Optional[str] = None,
         description: Optional[str] = None,
+        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
+        response_description: Optional[str] = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
         tags: Optional[List["Tag"]] = None,
+        operation_id: Optional[str] = None,
         middlewares: Optional[List[Callable[..., Any]]] = None,
     ):
         raise NotImplementedError()
@@ -649,9 +669,13 @@ def get(
         cors: Optional[bool] = None,
         compress: bool = False,
         cache_control: Optional[str] = None,
-        middlewares: Optional[List[Callable[..., Any]]] = None,
+        summary: Optional[str] = None,
         description: Optional[str] = None,
+        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
+        response_description: Optional[str] = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
         tags: Optional[List["Tag"]] = None,
+        operation_id: Optional[str] = None,
+        middlewares: Optional[List[Callable[..., Any]]] = None,
     ):
         """Get route decorator with GET `method`
 
@@ -675,17 +699,34 @@ def lambda_handler(event, context):
             return app.resolve(event, context)
         ```
         """
-        return self.route(rule, "GET", cors, compress, cache_control, description, tags, middlewares)
+        return self.route(
+            rule,
+            "GET",
+            cors,
+            compress,
+            cache_control,
+            summary,
+            description,
+            responses,
+            response_description,
+            tags,
+            operation_id,
+            middlewares,
+        )
 
     def post(
         self,
         rule: str,
         cors: Optional[bool] = None,
         compress: bool = False,
         cache_control: Optional[str] = None,
-        middlewares: Optional[List[Callable[..., Any]]] = None,
+        summary: Optional[str] = None,
         description: Optional[str] = None,
+        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
+        response_description: Optional[str] = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
         tags: Optional[List["Tag"]] = None,
+        operation_id: Optional[str] = None,
+        middlewares: Optional[List[Callable[..., Any]]] = None,
     ):
         """Post route decorator with POST `method`
 
@@ -710,17 +751,34 @@ def lambda_handler(event, context):
             return app.resolve(event, context)
         ```
         """
-        return self.route(rule, "POST", cors, compress, cache_control, description, tags, middlewares)
+        return self.route(
+            rule,
+            "POST",
+            cors,
+            compress,
+            cache_control,
+            summary,
+            description,
+            responses,
+            response_description,
+            tags,
+            operation_id,
+            middlewares,
+        )
 
     def put(
         self,
         rule: str,
         cors: Optional[bool] = None,
         compress: bool = False,
         cache_control: Optional[str] = None,
-        middlewares: Optional[List[Callable[..., Any]]] = None,
+        summary: Optional[str] = None,
         description: Optional[str] = None,
+        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
+        response_description: Optional[str] = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
         tags: Optional[List["Tag"]] = None,
+        operation_id: Optional[str] = None,
+        middlewares: Optional[List[Callable[..., Any]]] = None,
     ):
         """Put route decorator with PUT `method`
 
@@ -745,17 +803,34 @@ def lambda_handler(event, context):
             return app.resolve(event, context)
         ```
         """
-        return self.route(rule, "PUT", cors, compress, cache_control, description, tags, middlewares)
+        return self.route(
+            rule,
+            "PUT",
+            cors,
+            compress,
+            cache_control,
+            summary,
+            description,
+            responses,
+            response_description,
+            tags,
+            operation_id,
+            middlewares,
+        )
 
     def delete(
         self,
         rule: str,
         cors: Optional[bool] = None,
         compress: bool = False,
         cache_control: Optional[str] = None,
-        middlewares: Optional[List[Callable[..., Any]]] = None,
+        summary: Optional[str] = None,
         description: Optional[str] = None,
+        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
+        response_description: Optional[str] = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
         tags: Optional[List["Tag"]] = None,
+        operation_id: Optional[str] = None,
+        middlewares: Optional[List[Callable[..., Any]]] = None,
     ):
         """Delete route decorator with DELETE `method`
 
@@ -779,17 +854,34 @@ def lambda_handler(event, context):
             return app.resolve(event, context)
         ```
         """
-        return self.route(rule, "DELETE", cors, compress, cache_control, description, tags, middlewares)
+        return self.route(
+            rule,
+            "DELETE",
+            cors,
+            compress,
+            cache_control,
+            summary,
+            description,
+            responses,
+            response_description,
+            tags,
+            operation_id,
+            middlewares,
+        )
 
     def patch(
         self,
         rule: str,
         cors: Optional[bool] = None,
         compress: bool = False,
         cache_control: Optional[str] = None,
-        middlewares: Optional[List[Callable]] = None,
+        summary: Optional[str] = None,
         description: Optional[str] = None,
+        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
+        response_description: Optional[str] = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
         tags: Optional[List["Tag"]] = None,
+        operation_id: Optional[str] = None,
+        middlewares: Optional[List[Callable]] = None,
     ):
         """Patch route decorator with PATCH `method`
 
@@ -816,7 +908,20 @@ def lambda_handler(event, context):
             return app.resolve(event, context)
         ```
         """
-        return self.route(rule, "PATCH", cors, compress, cache_control, description, tags, middlewares)
+        return self.route(
+            rule,
+            "PATCH",
+            cors,
+            compress,
+            cache_control,
+            summary,
+            description,
+            responses,
+            response_description,
+            tags,
+            operation_id,
+            middlewares,
+        )
 
     def _push_processed_stack_frame(self, frame: str):
         """
@@ -1202,8 +1307,12 @@ def route(
         cors: Optional[bool] = None,
         compress: bool = False,
         cache_control: Optional[str] = None,
+        summary: Optional[str] = None,
         description: Optional[str] = None,
+        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
+        response_description: Optional[str] = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
         tags: Optional[List["Tag"]] = None,
+        operation_id: Optional[str] = None,
         middlewares: Optional[List[Callable[..., Any]]] = None,
     ):
         """Route decorator includes parameter `method`"""
@@ -1223,9 +1332,13 @@ def register_resolver(func: Callable):
                     cors_enabled,
                     compress,
                     cache_control,
-                    middlewares,
+                    summary,
                     description,
+                    responses,
+                    response_description,
                     tags,
+                    operation_id,
+                    middlewares,
                 )
 
                 # The more specific route wins.
@@ -1623,15 +1736,31 @@ def route(
         cors: Optional[bool] = None,
         compress: bool = False,
         cache_control: Optional[str] = None,
+        summary: Optional[str] = None,
         description: Optional[str] = None,
+        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
+        response_description: Optional[str] = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
         tags: Optional[List["Tag"]] = None,
+        operation_id: Optional[str] = None,
         middlewares: Optional[List[Callable[..., Any]]] = None,
     ):
         def register_route(func: Callable):
             # Convert methods to tuple. It needs to be hashable as its part of the self._routes dict key
             methods = (method,) if isinstance(method, str) else tuple(method)
 
-            route_key = (rule, methods, cors, compress, cache_control, description, tags)
+            route_key = (
+                rule,
+                methods,
+                cors,
+                compress,
+                cache_control,
+                summary,
+                description,
+                responses,
+                response_description,
+                tags,
+                operation_id,
+            )
 
             # Collate Middleware for routes
             if middlewares is not None:
@@ -1671,12 +1800,29 @@ def route(
         cors: Optional[bool] = None,
         compress: bool = False,
         cache_control: Optional[str] = None,
+        summary: Optional[str] = None,
         description: Optional[str] = None,
+        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
+        response_description: Optional[str] = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
         tags: Optional[List["Tag"]] = None,
+        operation_id: Optional[str] = None,
         middlewares: Optional[List[Callable[..., Any]]] = None,
     ):
         # NOTE: see #1552 for more context.
-        return super().route(rule.rstrip("/"), method, cors, compress, cache_control, description, tags, middlewares)
+        return super().route(
+            rule.rstrip("/"),
+            method,
+            cors,
+            compress,
+            cache_control,
+            summary,
+            description,
+            responses,
+            response_description,
+            tags,
+            operation_id,
+            middlewares,
+        )
 
     # Override _compile_regex to exclude trailing slashes for route resolution
     @staticmethod

aws_lambda_powertools/event_handler/openapi/params.py

@@ -351,7 +351,7 @@ def _get_field_info_and_type_annotation(annotation, value, is_path_param: bool)
     if annotation is not inspect.Signature.empty:
         # If the annotation is an Annotated type, we need to extract the type annotation and the FieldInfo
         if get_origin(annotation) is Annotated:
-            type_annotation = _get_field_info_annotated_type(annotation, value, is_path_param)
+            field_info, type_annotation = _get_field_info_annotated_type(annotation, value, is_path_param)
         # If the annotation is not an Annotated type, we use it as the type annotation
         else:
             type_annotation = annotation