pinecone.openapi_support.endpoint_utils

  1from .model_utils import file_type
  2from .exceptions import PineconeApiTypeError, PineconeApiValueError
  3from typing import Optional, Dict, Tuple, TypedDict, List, Literal, Any
  4from .types import PropertyValidationTypedDict
  5from .configuration import Configuration
  6from .model_utils import validate_and_convert_types, check_allowed_values, check_validations
  7
  8
  9class ExtraOpenApiKwargsTypedDict(TypedDict, total=False):
 10    _return_http_data_only: Optional[bool]
 11    _preload_content: Optional[bool]
 12    _request_timeout: Optional[int]
 13    _check_input_type: Optional[bool]
 14    _check_return_type: Optional[bool]
 15    async_req: Optional[bool]
 16
 17
 18class KwargsWithOpenApiKwargDefaultsTypedDict(TypedDict, total=False):
 19    _return_http_data_only: bool
 20    _preload_content: bool
 21    _request_timeout: int
 22    _check_input_type: bool
 23    _check_return_type: bool
 24    async_req: bool
 25
 26
 27class EndpointSettingsDict(TypedDict):
 28    response_type: Optional[Tuple]
 29    auth: List[str]
 30    endpoint_path: str
 31    operation_id: str
 32    http_method: Literal["POST", "PUT", "PATCH", "GET", "DELETE"]
 33    servers: Optional[List[str]]
 34
 35
 36class EndpointParamsMapDict(TypedDict):
 37    all: List[str]
 38    required: List[str]
 39    nullable: List[str]
 40    enum: List[str]
 41    validation: List[str]
 42
 43
 44AllowedValuesDict = Dict[Tuple[str], Dict]
 45
 46AttributeMapDictType = Dict[str, str]
 47LocationMapDictType = Dict[str, str]
 48OpenapiTypesDictType = Dict[str, Tuple]
 49
 50
 51class EndpointRootMapDict(TypedDict):
 52    validations: Dict[Tuple[str], PropertyValidationTypedDict]
 53    allowed_values: Dict[Tuple[str], Dict]
 54    openapi_types: OpenapiTypesDictType
 55    attribute_map: AttributeMapDictType
 56    location_map: LocationMapDictType
 57    collection_format_map: Dict[str, str]
 58
 59
 60class CombinedParamsMapDict(TypedDict):
 61    body: Any
 62    collection_format: Dict[str, str]
 63    file: Dict[str, List[file_type]]
 64    form: List[Tuple[str, Any]]
 65    header: Dict[str, List[str]]
 66    path: Dict[str, Any]
 67    query: List[Tuple[str, Any]]
 68
 69
 70class EndpointUtils:
 71    @staticmethod
 72    def gather_params(
 73        attribute_map: AttributeMapDictType,
 74        location_map: LocationMapDictType,
 75        openapi_types: OpenapiTypesDictType,
 76        collection_format_map: Dict[str, str],
 77        kwargs: Dict[str, Any],
 78    ) -> CombinedParamsMapDict:
 79        params: CombinedParamsMapDict = {
 80            "body": None,
 81            "collection_format": {},
 82            "file": {},
 83            "form": [],
 84            "header": {},
 85            "path": {},
 86            "query": [],
 87        }
 88
 89        for param_name, param_value in kwargs.items():
 90            param_location = location_map.get(param_name)
 91            if param_location is None:
 92                continue
 93            if param_location:
 94                if param_location == "body":
 95                    params["body"] = param_value
 96                    continue
 97                base_name = attribute_map[param_name]
 98                if param_location == "form" and openapi_types[param_name] == (file_type,):
 99                    params["file"][param_name] = [param_value]
100                elif param_location == "form" and openapi_types[param_name] == ([file_type],):
101                    # param_value is already a list
102                    params["file"][param_name] = param_value
103                elif param_location == "form":
104                    param_value_full = (base_name, param_value)
105                    params["form"].append(param_value_full)
106                elif param_location == "query":
107                    param_value_full = (base_name, param_value)
108                    params["query"].append(param_value_full)
109                elif param_location == "header":
110                    params["header"][base_name] = param_value
111                elif param_location == "path":
112                    params["path"][base_name] = param_value
113                else:
114                    raise PineconeApiTypeError("Got an unexpected location '%s' for parameter `%s`")
115
116                collection_format = collection_format_map.get(param_name)
117                if collection_format:
118                    params["collection_format"][base_name] = collection_format
119
120        return params
121
122    @staticmethod
123    def raise_if_missing_required_params(
124        params_map: EndpointParamsMapDict, settings: EndpointSettingsDict, kwargs: Dict[str, Any]
125    ) -> None:
126        for key in params_map["required"]:
127            if key not in kwargs.keys():
128                raise PineconeApiValueError(
129                    "Missing the required parameter `%s` when calling "
130                    "`%s`" % (key, settings["operation_id"])
131                )
132
133    @staticmethod
134    def raise_if_unexpected_param(
135        params_map: EndpointParamsMapDict, settings: EndpointSettingsDict, kwargs: Dict[str, Any]
136    ) -> None:
137        for key, value in kwargs.items():
138            if key not in params_map["all"]:
139                raise PineconeApiTypeError(
140                    "Got an unexpected parameter '%s'"
141                    " to method `%s`" % (key, settings["operation_id"])
142                )
143            # only throw this nullable PineconeApiValueError if _check_input_type
144            # is False, if _check_input_type==True we catch this case
145            # in self.__validate_inputs
146            if (
147                key not in params_map["nullable"]
148                and value is None
149                and kwargs["_check_input_type"] is False
150            ):
151                raise PineconeApiValueError(
152                    "Value may not be None for non-nullable parameter `%s`"
153                    " when calling `%s`" % (key, settings["operation_id"])
154                )
155
156    @staticmethod
157    def raise_if_invalid_inputs(
158        config: Configuration,
159        params_map: EndpointParamsMapDict,
160        allowed_values: AllowedValuesDict,
161        validations: PropertyValidationTypedDict,
162        openapi_types: OpenapiTypesDictType,
163        kwargs: Dict[str, Any],
164    ) -> None:
165        for param in params_map["enum"]:
166            if param in kwargs:
167                check_allowed_values(allowed_values, (param,), kwargs[param])
168
169        for param in params_map["validation"]:
170            if param in kwargs:
171                check_validations(validations, (param,), kwargs[param], configuration=config)
172
173        if kwargs["_check_input_type"] is False:
174            return
175
176        for key, value in kwargs.items():
177            fixed_val = validate_and_convert_types(
178                value,
179                openapi_types[key],
180                [key],
181                False,
182                kwargs["_check_input_type"],
183                configuration=config,
184            )
185            kwargs[key] = fixed_val