diff --git a/tests/e2e/event_handler/handlers/openapi_handler_with_pep563.py b/tests/e2e/event_handler/handlers/openapi_handler_with_pep563.py new file mode 100644 index 00000000000..a6f0ba29a8b --- /dev/null +++ b/tests/e2e/event_handler/handlers/openapi_handler_with_pep563.py @@ -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) diff --git a/tests/e2e/event_handler/infrastructure.py b/tests/e2e/event_handler/infrastructure.py index 9d7dbc46c40..142034e89b2 100644 --- a/tests/e2e/event_handler/infrastructure.py +++ b/tests/e2e/event_handler/infrastructure.py @@ -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): diff --git a/tests/e2e/event_handler/test_openapi.py b/tests/e2e/event_handler/test_openapi.py index d69c3b142b2..3a8d6b3c008 100644 --- a/tests/e2e/event_handler/test_openapi.py +++ b/tests/e2e/event_handler/test_openapi.py @@ -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 diff --git a/tests/functional/event_handler/_pydantic/test_openapi_with_pep563.py b/tests/functional/event_handler/_pydantic/test_openapi_with_pep563.py new file mode 100644 index 00000000000..1855aef45e2 --- /dev/null +++ b/tests/functional/event_handler/_pydantic/test_openapi_with_pep563.py @@ -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]