Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion aws_lambda_powertools/event_handler/openapi/params.py
Original file line number Diff line number Diff line change
Expand Up @@ -901,7 +901,7 @@ def analyze_param(
if is_response_param:
field_info.default = Required

field = _create_model_field(field_info, type_annotation, param_name, is_path_param)
field = _create_model_field(field_info, type_annotation, param_name, is_path_param, is_response_param)
return field


Expand Down Expand Up @@ -1138,6 +1138,7 @@ def _create_model_field(
type_annotation: Any,
param_name: str,
is_path_param: bool,
is_response_param: bool = False,
) -> ModelField | None:
"""
Create a new ModelField from a FieldInfo and type annotation.
Expand All @@ -1164,4 +1165,5 @@ def _create_model_field(
alias=field_info.alias,
required=field_info.default in (Required, Undefined),
field_info=field_info,
mode="serialization" if is_response_param else "validation",
)
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from typing import Literal, Optional

import pytest
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, computed_field
from typing_extensions import Annotated

from aws_lambda_powertools.event_handler import APIGatewayRestResolver
Expand Down Expand Up @@ -110,3 +110,79 @@ def create_todo(todo: TodoEnvelope): ...

# THEN the schema should be valid
assert openapi31_schema(schema)


@pytest.mark.usefixtures("pydanticv2_only")
def test_openapi_schema_includes_computed_field():
# GIVEN a model with a computed_field
class User(BaseModel):
first_name: str
last_name: str

@computed_field
@property
def full_name(self) -> str:
return f"{self.first_name} {self.last_name}"

# GIVEN APIGatewayRestResolver with a handler returning that model
app = APIGatewayRestResolver(enable_validation=True)

@app.get("/user")
def get_user() -> User:
return User(first_name="John", last_name="Doe")

# WHEN we get the schema
schema = json.loads(app.get_openapi_json_schema())

# THEN the computed_field should appear in the response schema
user_schema = schema["components"]["schemas"]["User"]
assert "full_name" in user_schema["properties"]
assert user_schema["properties"]["full_name"]["type"] == "string"
assert user_schema["properties"]["full_name"].get("readOnly") is True


@pytest.mark.usefixtures("pydanticv2_only")
def test_openapi_schema_computed_field_not_in_request_body():
# GIVEN a model with a computed_field used as both request and response
class Item(BaseModel):
price: float
quantity: int

@computed_field
@property
def total(self) -> float:
return self.price * self.quantity

# GIVEN APIGatewayRestResolver with handlers using the model
app = APIGatewayRestResolver(enable_validation=True)

@app.post("/items")
def create_item(item: Item) -> Item:
return item

# WHEN we get the schema
schema = json.loads(app.get_openapi_json_schema())

# THEN the request body schema should NOT include computed_field
request_body = schema["paths"]["/items"]["post"]["requestBody"]
request_ref = request_body["content"]["application/json"]["schema"]["$ref"]
request_schema_name = request_ref.split("/")[-1]

# THEN the response schema SHOULD include computed_field
response_ref = schema["paths"]["/items"]["post"]["responses"]["200"]["content"]["application/json"]["schema"][
"$ref"
]
response_schema_name = response_ref.split("/")[-1]

# When input/output schemas are separate, we expect different schema names
# When they share a schema, computed_field should be present
if request_schema_name == response_schema_name:
# Shared schema - computed_field should be present (serialization mode wins)
item_schema = schema["components"]["schemas"][response_schema_name]
assert "total" in item_schema["properties"]
else:
# Separate schemas
input_schema = schema["components"]["schemas"][request_schema_name]
output_schema = schema["components"]["schemas"][response_schema_name]
assert "total" not in input_schema["properties"]
assert "total" in output_schema["properties"]
Loading