|
92 | 92 | Server,
|
93 | 93 | Tag,
|
94 | 94 | )
|
95 |
| - from aws_lambda_powertools.event_handler.openapi.params import Dependant |
| 95 | + from aws_lambda_powertools.event_handler.openapi.params import Dependant, Param |
96 | 96 | from aws_lambda_powertools.event_handler.openapi.swagger_ui.oauth2 import (
|
97 | 97 | OAuth2Config,
|
98 | 98 | )
|
@@ -812,74 +812,110 @@ def _openapi_operation_parameters(
|
812 | 812 | """
|
813 | 813 | Returns the OpenAPI operation parameters.
|
814 | 814 | """
|
815 |
| - from aws_lambda_powertools.event_handler.openapi.compat import get_schema_from_model_field |
816 | 815 | from aws_lambda_powertools.event_handler.openapi.params import Param
|
817 | 816 |
|
818 |
| - parameters = [] |
819 |
| - parameter: dict[str, Any] = {} |
| 817 | + parameters: list[dict[str, Any]] = [] |
820 | 818 |
|
821 | 819 | for param in all_route_params:
|
822 |
| - field_info = param.field_info |
823 |
| - field_info = cast(Param, field_info) |
| 820 | + field_info = cast(Param, param.field_info) |
824 | 821 | if not field_info.include_in_schema:
|
825 | 822 | continue
|
826 | 823 |
|
827 | 824 | # Check if this is a Pydantic model that should be expanded
|
828 |
| - from pydantic import BaseModel |
829 |
| - |
830 |
| - from aws_lambda_powertools.event_handler.openapi.compat import lenient_issubclass |
831 |
| - |
832 |
| - if lenient_issubclass(field_info.annotation, BaseModel): |
833 |
| - # Expand Pydantic model into individual parameters |
834 |
| - model_class = cast(type[BaseModel], field_info.annotation) |
835 |
| - |
836 |
| - for field_name, field_def in model_class.model_fields.items(): |
837 |
| - if not field_def.annotation: |
838 |
| - continue |
839 |
| - # Create individual parameter for each model field |
840 |
| - param_name = field_def.alias or field_name |
841 |
| - |
842 |
| - individual_param = { |
843 |
| - "name": param_name, |
844 |
| - "in": field_info.in_.value, |
845 |
| - "required": field_def.is_required() |
846 |
| - if hasattr(field_def, "is_required") |
847 |
| - else field_def.default is ..., |
848 |
| - "schema": Route._get_basic_type_schema(field_def.annotation), |
849 |
| - } |
| 825 | + if Route._is_pydantic_model_param(field_info): |
| 826 | + parameters.extend(Route._expand_pydantic_model_parameters(field_info)) |
| 827 | + else: |
| 828 | + parameters.append(Route._create_regular_parameter(param, model_name_map, field_mapping)) |
850 | 829 |
|
851 |
| - if field_def.description: |
852 |
| - individual_param["description"] = field_def.description |
| 830 | + return parameters |
853 | 831 |
|
854 |
| - parameters.append(individual_param) |
855 |
| - else: |
856 |
| - # Regular parameter processing |
857 |
| - param_schema = get_schema_from_model_field( |
858 |
| - field=param, |
859 |
| - model_name_map=model_name_map, |
860 |
| - field_mapping=field_mapping, |
861 |
| - ) |
| 832 | + @staticmethod |
| 833 | + def _is_pydantic_model_param(field_info: ModelField | Param) -> bool: |
| 834 | + """Check if the field info represents a Pydantic model parameter.""" |
| 835 | + from pydantic import BaseModel |
| 836 | + |
| 837 | + from aws_lambda_powertools.event_handler.openapi.compat import lenient_issubclass |
| 838 | + from aws_lambda_powertools.event_handler.openapi.params import Param |
862 | 839 |
|
863 |
| - parameter = { |
864 |
| - "name": param.alias, |
865 |
| - "in": field_info.in_.value, |
866 |
| - "required": param.required, |
867 |
| - "schema": param_schema, |
868 |
| - } |
| 840 | + if not isinstance(field_info, Param): |
| 841 | + return False |
| 842 | + return lenient_issubclass(field_info.annotation, BaseModel) |
869 | 843 |
|
870 |
| - if field_info.description: |
871 |
| - parameter["description"] = field_info.description |
| 844 | + @staticmethod |
| 845 | + def _expand_pydantic_model_parameters(field_info: Param) -> list[dict[str, Any]]: |
| 846 | + """Expand a Pydantic model into individual OpenAPI parameters.""" |
| 847 | + from pydantic import BaseModel |
872 | 848 |
|
873 |
| - if field_info.openapi_examples: |
874 |
| - parameter["examples"] = field_info.openapi_examples |
| 849 | + model_class = cast(type[BaseModel], field_info.annotation) |
| 850 | + parameters: list[dict[str, Any]] = [] |
875 | 851 |
|
876 |
| - if field_info.deprecated: |
877 |
| - parameter["deprecated"] = field_info.deprecated |
| 852 | + for field_name, field_def in model_class.model_fields.items(): |
| 853 | + if not field_def.annotation: |
| 854 | + continue |
878 | 855 |
|
879 |
| - parameters.append(parameter) |
| 856 | + param_name = field_def.alias or field_name |
| 857 | + individual_param = Route._create_pydantic_field_parameter( |
| 858 | + param_name=param_name, |
| 859 | + field_def=field_def, |
| 860 | + param_location=field_info.in_.value, |
| 861 | + ) |
| 862 | + parameters.append(individual_param) |
880 | 863 |
|
881 | 864 | return parameters
|
882 | 865 |
|
| 866 | + @staticmethod |
| 867 | + def _create_pydantic_field_parameter( |
| 868 | + param_name: str, |
| 869 | + field_def: Any, |
| 870 | + param_location: str, |
| 871 | + ) -> dict[str, Any]: |
| 872 | + """Create an OpenAPI parameter from a Pydantic field definition.""" |
| 873 | + individual_param: dict[str, Any] = { |
| 874 | + "name": param_name, |
| 875 | + "in": param_location, |
| 876 | + "required": field_def.is_required() if hasattr(field_def, "is_required") else field_def.default is ..., |
| 877 | + "schema": Route._get_basic_type_schema(field_def.annotation or type(None)), |
| 878 | + } |
| 879 | + |
| 880 | + if field_def.description: |
| 881 | + individual_param["description"] = field_def.description |
| 882 | + |
| 883 | + return individual_param |
| 884 | + |
| 885 | + @staticmethod |
| 886 | + def _create_regular_parameter( |
| 887 | + param: ModelField, |
| 888 | + model_name_map: dict[TypeModelOrEnum, str], |
| 889 | + field_mapping: dict[tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue], |
| 890 | + ) -> dict[str, Any]: |
| 891 | + """Create an OpenAPI parameter from a regular ModelField.""" |
| 892 | + from aws_lambda_powertools.event_handler.openapi.compat import get_schema_from_model_field |
| 893 | + from aws_lambda_powertools.event_handler.openapi.params import Param |
| 894 | + |
| 895 | + field_info = cast(Param, param.field_info) |
| 896 | + param_schema = get_schema_from_model_field( |
| 897 | + field=param, |
| 898 | + model_name_map=model_name_map, |
| 899 | + field_mapping=field_mapping, |
| 900 | + ) |
| 901 | + |
| 902 | + parameter: dict[str, Any] = { |
| 903 | + "name": param.alias, |
| 904 | + "in": field_info.in_.value, |
| 905 | + "required": param.required, |
| 906 | + "schema": param_schema, |
| 907 | + } |
| 908 | + |
| 909 | + # Add optional attributes if present |
| 910 | + if field_info.description: |
| 911 | + parameter["description"] = field_info.description |
| 912 | + if field_info.openapi_examples: |
| 913 | + parameter["examples"] = field_info.openapi_examples |
| 914 | + if field_info.deprecated: |
| 915 | + parameter["deprecated"] = field_info.deprecated |
| 916 | + |
| 917 | + return parameter |
| 918 | + |
883 | 919 | @staticmethod
|
884 | 920 | def _get_basic_type_schema(param_type: type) -> dict[str, str]:
|
885 | 921 | """
|
|
0 commit comments