Skip to content
Merged
Show file tree
Hide file tree
Changes from 9 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
16 changes: 11 additions & 5 deletions aws_lambda_powertools/event_handler/api_gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -661,30 +661,36 @@ def _get_openapi_path( # noqa PLR0912
else:
# Need to iterate to transform any 'model' into a 'schema'
for content_type, payload in response["content"].items():
new_payload: OpenAPIResponseContentSchema

# Case 2.1: the 'content' has a model
if "model" in payload:
# Find the model in the dependant's extra models
model_payload_typed = cast(OpenAPIResponseContentModel, payload)
return_field = next(
filter(
lambda model: model.type_ is cast(OpenAPIResponseContentModel, payload)["model"],
lambda model: model.type_ is model_payload_typed["model"],
self.dependant.response_extra_models,
),
)
if not return_field:
raise AssertionError("Model declared in custom responses was not found")

new_payload = self._openapi_operation_return(
model_payload = self._openapi_operation_return(
param=return_field,
model_name_map=model_name_map,
field_mapping=field_mapping,
)

# Preserve existing fields like examples, encoding, etc.
new_payload: OpenAPIResponseContentSchema = {}
for key, value in payload.items():
if key != "model":
new_payload[key] = value # type: ignore[literal-required]
new_payload.update(model_payload) # Add/override with model schema

# Case 2.2: the 'content' has a schema
else:
# Do nothing! We already have what we need!
new_payload = payload
new_payload = cast(OpenAPIResponseContentSchema, payload)

response["content"][content_type] = new_payload

Expand Down
24 changes: 21 additions & 3 deletions aws_lambda_powertools/event_handler/openapi/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,14 +63,32 @@
}


class OpenAPIResponseHeader(TypedDict, total=False):
"""OpenAPI Response Header Object"""

description: NotRequired[str]
schema: NotRequired[dict[str, Any]]
examples: NotRequired[dict[str, Any]]
style: NotRequired[str]
explode: NotRequired[bool]
allowReserved: NotRequired[bool]
deprecated: NotRequired[bool]


class OpenAPIResponseContentSchema(TypedDict, total=False):
schema: dict
examples: NotRequired[dict[str, Any]]
encoding: NotRequired[dict[str, Any]]


class OpenAPIResponseContentModel(TypedDict):
class OpenAPIResponseContentModel(TypedDict, total=False):
model: Any
examples: NotRequired[dict[str, Any]]
encoding: NotRequired[dict[str, Any]]


class OpenAPIResponse(TypedDict):
description: str
class OpenAPIResponse(TypedDict, total=False):
description: str # Still required
headers: NotRequired[dict[str, OpenAPIResponseHeader]]
content: NotRequired[dict[str, OpenAPIResponseContentSchema | OpenAPIResponseContentModel]]
links: NotRequired[dict[str, Any]]
Loading