aws-powertools · leandrodamascena · Jan 20, 2025 · Jan 20, 2025
@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from aws_lambda_powertools.event_handler import (
+    APIGatewayRestResolver,
+)
+
+
+class Todo(BaseModel):
+    id: int = Field(examples=[1])
+    title: str = Field(examples=["Example 1"])
+    priority: float = Field(examples=[0.5])
+    completed: bool = Field(examples=[True])
+
+
+app = APIGatewayRestResolver(enable_validation=True)
+
+
+@app.get("/openapi_schema_with_pep563")
+def openapi_schema():
+    return app.get_openapi_json_schema(
+        title="Powertools e2e API",
+        version="1.0.0",
+        description="This is a sample Powertools e2e API",
+        openapi_extensions={"x-amazon-apigateway-gateway-responses": {"DEFAULT_4XX"}},
+    )
+
+
+@app.get("/")
+def handler() -> Todo:
+    return Todo(id=0, title="", priority=0.0, completed=False)
+
+
+def lambda_handler(event, context):
+    return app.resolve(event, context)
@@ -18,7 +18,13 @@ def create_resources(self):
         functions = self.create_lambda_functions(function_props={"timeout": Duration.seconds(10)})
 
         self._create_alb(function=[functions["AlbHandler"], functions["AlbHandlerWithBodyNone"]])
-        self._create_api_gateway_rest(function=[functions["ApiGatewayRestHandler"], functions["OpenapiHandler"]])
+        self._create_api_gateway_rest(
+            function=[
+                functions["ApiGatewayRestHandler"],
+                functions["OpenapiHandler"],
+                functions["OpenapiHandlerWithPep563"],
+            ],
+        )
         self._create_api_gateway_http(function=functions["ApiGatewayHttpHandler"])
         self._create_lambda_function_url(function=functions["LambdaFunctionUrlHandler"])
 
@@ -92,6 +98,9 @@ def _create_api_gateway_rest(self, function: List[Function]):
         openapi_schema = apigw.root.add_resource("openapi_schema")
         openapi_schema.add_method("GET", apigwv1.LambdaIntegration(function[1], proxy=True))
 
+        openapi_schema = apigw.root.add_resource("openapi_schema_with_pep563")
+        openapi_schema.add_method("GET", apigwv1.LambdaIntegration(function[2], proxy=True))
+
         CfnOutput(self.stack, "APIGatewayRestUrl", value=apigw.url)
 
     def _create_lambda_function_url(self, function: Function):

@@ -25,3 +25,20 @@ def test_get_openapi_schema(apigw_rest_endpoint):
     assert "Powertools e2e API" in response.text
     assert "x-amazon-apigateway-gateway-responses" in response.text
     assert response.status_code == 200
+
+
+def test_get_openapi_schema_with_pep563(apigw_rest_endpoint):
+    # GIVEN
+    url = f"{apigw_rest_endpoint}openapi_schema_with_pep563"
+
+    # WHEN
+    response = data_fetcher.get_http_response(
+        Request(
+            method="GET",
+            url=url,
+        ),
+    )
+
+    assert "Powertools e2e API" in response.text
+    assert "x-amazon-apigateway-gateway-responses" in response.text
+    assert response.status_code == 200
@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+from typing_extensions import Annotated
+
+from aws_lambda_powertools.event_handler.api_gateway import APIGatewayRestResolver
+from aws_lambda_powertools.event_handler.openapi.models import (
+    ParameterInType,
+    Schema,
+)
+from aws_lambda_powertools.event_handler.openapi.params import (
+    Body,
+    Query,
+)
+
+JSON_CONTENT_TYPE = "application/json"
+
+
+class Todo(BaseModel):
+    id: int = Field(examples=[1])
+    title: str = Field(examples=["Example 1"])
+    priority: float = Field(examples=[0.5])
+    completed: bool = Field(examples=[True])
+
+
+def test_openapi_with_pep563_and_input_model():
+    app = APIGatewayRestResolver()
+
+    @app.get("/users", summary="Get Users", operation_id="GetUsers", description="Get paginated users", tags=["Users"])
+    def handler(
+        count: Annotated[
+            int,
+            Query(gt=0, lt=100, examples=["Example 1"]),
+        ] = 1,
+    ):
+        print(count)
+        raise NotImplementedError()
+
+    schema = app.get_openapi_schema()
+
+    get = schema.paths["/users"].get
+    assert len(get.parameters) == 1
+    assert get.summary == "Get Users"
+    assert get.operationId == "GetUsers"
+    assert get.description == "Get paginated users"
+    assert get.tags == ["Users"]
+
+    parameter = get.parameters[0]
+    assert parameter.required is False
+    assert parameter.name == "count"
+    assert parameter.in_ == ParameterInType.query
+    assert parameter.schema_.type == "integer"
+    assert parameter.schema_.default == 1
+    assert parameter.schema_.title == "Count"
+    assert parameter.schema_.exclusiveMinimum == 0
+    assert parameter.schema_.exclusiveMaximum == 100
+    assert len(parameter.schema_.examples) == 1
+    assert parameter.schema_.examples[0] == "Example 1"
+
+
+def test_openapi_with_pep563_and_output_model():
+
+    app = APIGatewayRestResolver()
+
+    @app.get("/")
+    def handler() -> Todo:
+        return Todo(id=0, title="", priority=0.0, completed=False)
+
+    schema = app.get_openapi_schema()
+    assert "Todo" in schema.components.schemas
+    todo_schema = schema.components.schemas["Todo"]
+    assert isinstance(todo_schema, Schema)
+
+    assert "id" in todo_schema.properties
+    id_property = todo_schema.properties["id"]
+    assert id_property.examples == [1]
+
+    assert "title" in todo_schema.properties
+    title_property = todo_schema.properties["title"]
+    assert title_property.examples == ["Example 1"]
+
+    assert "priority" in todo_schema.properties
+    priority_property = todo_schema.properties["priority"]
+    assert priority_property.examples == [0.5]
+
+    assert "completed" in todo_schema.properties
+    completed_property = todo_schema.properties["completed"]
+    assert completed_property.examples == [True]
+
+
+def test_openapi_with_pep563_and_annotated_body():
+
+    app = APIGatewayRestResolver()
+
+    @app.post("/todo")
+    def create_todo(
+        todo_create_request: Annotated[Todo, Body(title="New Todo")],
+    ) -> dict:
+        return {"message": f"Created todo {todo_create_request.title}"}
+
+    schema = app.get_openapi_schema()
+    assert "Todo" in schema.components.schemas
+    todo_schema = schema.components.schemas["Todo"]
+    assert isinstance(todo_schema, Schema)
+
+    assert "id" in todo_schema.properties
+    id_property = todo_schema.properties["id"]
+    assert id_property.examples == [1]
+
+    assert "title" in todo_schema.properties
+    title_property = todo_schema.properties["title"]
+    assert title_property.examples == ["Example 1"]
+
+    assert "priority" in todo_schema.properties
+    priority_property = todo_schema.properties["priority"]
+    assert priority_property.examples == [0.5]
+
+    assert "completed" in todo_schema.properties
+    completed_property = todo_schema.properties["completed"]
+    assert completed_property.examples == [True]