pinecone.openapi_support.api_client_utils

  1import json
  2import mimetypes
  3import io
  4import os
  5from urllib3.fields import RequestField
  6from urllib.parse import quote
  7
  8from typing import Optional, List, Tuple, Dict, Any, Union
  9from .serializer import Serializer
 10from .exceptions import PineconeApiValueError
 11
 12
 13class HeaderUtil:
 14    @staticmethod
 15    def select_header_content_type(content_types: List[str]) -> str:
 16        """Returns `Content-Type` based on an array of content_types provided.
 17
 18        :param content_types: List of content-types.
 19        :return: Content-Type (e.g. application/json).
 20        """
 21        if not content_types:
 22            return "application/json"
 23
 24        content_types = [x.lower() for x in content_types]
 25
 26        if "application/json" in content_types or "*/*" in content_types:
 27            return "application/json"
 28        else:
 29            return content_types[0]
 30
 31    @staticmethod
 32    def select_header_accept(accepts: List[str]) -> str:
 33        """Returns `Accept` based on an array of accepts provided.
 34
 35        :param accepts: List of headers.
 36        :return: Accept (e.g. application/json).
 37        """
 38        if not accepts:
 39            return ""
 40
 41        accepts = [x.lower() for x in accepts]
 42
 43        if "application/json" in accepts:
 44            return "application/json"
 45        else:
 46            return ", ".join(accepts)
 47
 48    @staticmethod
 49    def process_header_params(
 50        default_headers: Dict[str, str], header_params: Dict[str, str], collection_formats
 51    ) -> Dict[str, Any]:
 52        header_params.update(default_headers)
 53        if header_params:
 54            sanitized_header_params: Dict[str, Any] = Serializer.sanitize_for_serialization(
 55                header_params
 56            )
 57            processed_header_params: Dict[str, Any] = dict(
 58                parameters_to_tuples(sanitized_header_params, collection_formats)
 59            )
 60        return processed_header_params
 61
 62    @staticmethod
 63    def prepare_headers(headers_map: Dict[str, List[str]], params) -> None:
 64        """Mutates the params to set Accept and Content-Type headers."""
 65        accept_headers_list = headers_map["accept"]
 66        if accept_headers_list:
 67            params["header"]["Accept"] = HeaderUtil.select_header_accept(accept_headers_list)
 68
 69        content_type_headers_list = headers_map["content_type"]
 70        if content_type_headers_list:
 71            header_list = HeaderUtil.select_header_content_type(content_type_headers_list)
 72            params["header"]["Content-Type"] = header_list
 73
 74
 75def process_query_params(query_params, collection_formats):
 76    if query_params:
 77        sanitized_query_params = Serializer.sanitize_for_serialization(query_params)
 78        processed_query_params = parameters_to_tuples(sanitized_query_params, collection_formats)
 79    else:
 80        processed_query_params = []
 81
 82    return processed_query_params
 83
 84
 85def process_params(
 86    default_headers: Dict[str, str],
 87    header_params: Dict[str, Any],
 88    path_params: Dict[str, Any],
 89    collection_formats: Dict[str, str],
 90):
 91    # header parameters
 92    headers_tuple = HeaderUtil.process_header_params(
 93        default_headers, header_params, collection_formats
 94    )
 95
 96    # path parameters
 97    sanitized_path_params: Dict[str, Any] = Serializer.sanitize_for_serialization(path_params or {})
 98    path_parm: List[Tuple[str, Any]] = parameters_to_tuples(
 99        sanitized_path_params, collection_formats
100    )
101
102    return headers_tuple, path_parm, sanitized_path_params
103
104
105def parameters_to_multipart(params, collection_types):
106    """Get parameters as list of tuples, formatting as json if value is collection_types
107
108    :param params: Parameters as list of two-tuples
109    :param dict collection_types: Parameter collection types
110    :return: Parameters as list of tuple or urllib3.fields.RequestField
111    """
112    new_params = []
113    if collection_types is None:
114        collection_types = dict
115    for k, v in params.items() if isinstance(params, dict) else params:  # noqa: E501
116        if isinstance(
117            v, collection_types
118        ):  # v is instance of collection_type, formatting as application/json
119            v = json.dumps(v, ensure_ascii=False).encode("utf-8")
120            field = RequestField(k, v)
121            field.make_multipart(content_type="application/json; charset=utf-8")
122            new_params.append(field)
123        else:
124            new_params.append((k, v))
125    return new_params
126
127
128def files_parameters(files: Optional[Dict[str, List[io.IOBase]]] = None):
129    """Builds form parameters.
130
131    :param files: None or a dict with key=param_name and
132        value is a list of open file objects
133    :return: List of tuples of form parameters with file data
134    """
135    if files is None:
136        return []
137
138    params = []
139    for param_name, file_instances in files.items():
140        if file_instances is None:
141            # if the file field is nullable, skip None values
142            continue
143        for file_instance in file_instances:
144            if file_instance is None:
145                # if the file field is nullable, skip None values
146                continue
147            if file_instance.closed is True:
148                raise PineconeApiValueError(
149                    "Cannot read a closed file. The passed in file_type "
150                    "for %s must be open." % param_name
151                )
152            filename = os.path.basename(file_instance.name)  # type: ignore
153            filedata = Serializer.get_file_data_and_close_file(file_instance)
154            mimetype = mimetypes.guess_type(filename)[0] or "application/octet-stream"
155            params.append(tuple([param_name, tuple([filename, filedata, mimetype])]))
156
157    return params
158
159
160def parameters_to_tuples(
161    params: Union[Dict[str, Any], List[Tuple[str, Any]]],
162    collection_formats: Optional[Dict[str, str]],
163) -> List[Tuple[str, str]]:
164    """Get parameters as list of tuples, formatting collections.
165
166    :param params: Parameters as dict or list of two-tuples
167    :param dict collection_formats: Parameter collection formats
168    :return: Parameters as list of tuples, collections formatted
169    """
170    new_params: List[Tuple[str, Any]] = []
171    if collection_formats is None:
172        collection_formats = {}
173    for k, v in params.items() if isinstance(params, dict) else params:  # noqa: E501
174        if k in collection_formats:
175            collection_format = collection_formats[k]
176            if collection_format == "multi":
177                new_params.extend((k, value) for value in v)
178            else:
179                if collection_format == "ssv":
180                    delimiter = " "
181                elif collection_format == "tsv":
182                    delimiter = "\t"
183                elif collection_format == "pipes":
184                    delimiter = "|"
185                else:  # csv is the default
186                    delimiter = ","
187                new_params.append((k, delimiter.join(str(value) for value in v)))
188        else:
189            new_params.append((k, v))
190    return new_params
191
192
193def build_request_url(config, processed_path_params, resource_path, _host):
194    for k, v in processed_path_params:
195        # specified safe chars, encode everything
196        resource_path = resource_path.replace(
197            "{%s}" % k, quote(str(v), safe=config.safe_chars_for_path_param)
198        )
199
200    # _host is a host override passed for an individual operation
201    host = _host if _host is not None else config.host
202
203    return host + resource_path