use Mapping instead of Dict, Sequence instead of List

Author: Daniel Melo

pinecone/core/client/api_client.py

@@ -118,16 +118,16 @@ def __call_api(
         self,
         resource_path: str,
         method: str,
-        path_params: typing.Optional[typing.Dict[str, typing.Any]] = None,
-        query_params: typing.Optional[typing.List[typing.Tuple[str, typing.Any]]] = None,
-        header_params: typing.Optional[typing.Dict[str, typing.Any]] = None,
+        path_params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
+        query_params: typing.Optional[typing.Sequence[typing.Tuple[str, typing.Any]]] = None,
+        header_params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
         body: typing.Optional[typing.Any] = None,
-        post_params: typing.Optional[typing.List[typing.Tuple[str, typing.Any]]] = None,
-        files: typing.Optional[typing.Dict[str, typing.List[io.IOBase]]] = None,
+        post_params: typing.Optional[typing.Sequence[typing.Tuple[str, typing.Any]]] = None,
+        files: typing.Optional[typing.Mapping[str, typing.Sequence[io.IOBase]]] = None,
         response_type: typing.Optional[typing.Tuple[typing.Any]] = None,
-        auth_settings: typing.Optional[typing.List[str]] = None,
+        auth_settings: typing.Optional[typing.Sequence[str]] = None,
         _return_http_data_only: typing.Optional[bool] = None,
-        collection_formats: typing.Optional[typing.Dict[str, str]] = None,
+        collection_formats: typing.Optional[typing.Mapping[str, str]] = None,
         _preload_content: bool = True,
         _request_timeout: typing.Optional[typing.Union[int, float, typing.Tuple]] = None,
         _host: typing.Optional[str] = None,
@@ -280,9 +280,9 @@ def sanitize_for_serialization(cls, obj):
             return obj.isoformat()
         elif isinstance(obj, ModelSimple):
             return cls.sanitize_for_serialization(obj.value)
-        elif isinstance(obj, (list, tuple)):
+        elif isinstance(obj, typing.Sequence):
             return [cls.sanitize_for_serialization(item) for item in obj]
-        if isinstance(obj, dict):
+        if isinstance(obj, typing.Mapping):
             return {key: cls.sanitize_for_serialization(val) for key, val in obj.items()}
         raise PineconeApiValueError('Unable to prepare type {} for serialization'.format(obj.__class__.__name__))
 
@@ -335,17 +335,17 @@ def call_api(
         self,
         resource_path: str,
         method: str,
-        path_params: typing.Optional[typing.Dict[str, typing.Any]] = None,
-        query_params: typing.Optional[typing.List[typing.Tuple[str, typing.Any]]] = None,
-        header_params: typing.Optional[typing.Dict[str, typing.Any]] = None,
+        path_params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
+        query_params: typing.Optional[typing.Sequence[typing.Tuple[str, typing.Any]]] = None,
+        header_params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
         body: typing.Optional[typing.Any] = None,
-        post_params: typing.Optional[typing.List[typing.Tuple[str, typing.Any]]] = None,
-        files: typing.Optional[typing.Dict[str, typing.List[io.IOBase]]] = None,
+        post_params: typing.Optional[typing.Sequence[typing.Tuple[str, typing.Any]]] = None,
+        files: typing.Optional[typing.Mapping[str, typing.Sequence[io.IOBase]]] = None,
         response_type: typing.Optional[typing.Tuple[typing.Any]] = None,
-        auth_settings: typing.Optional[typing.List[str]] = None,
+        auth_settings: typing.Optional[typing.Sequence[str]] = None,
         async_req: typing.Optional[bool] = None,
         _return_http_data_only: typing.Optional[bool] = None,
-        collection_formats: typing.Optional[typing.Dict[str, str]] = None,
+        collection_formats: typing.Optional[typing.Mapping[str, str]] = None,
         _preload_content: bool = True,
         _request_timeout: typing.Optional[typing.Union[int, float, typing.Tuple]] = None,
         _host: typing.Optional[str] = None,
@@ -497,7 +497,7 @@ def parameters_to_tuples(self, params, collection_formats):
         new_params = []
         if collection_formats is None:
             collection_formats = {}
-        for k, v in params.items() if isinstance(params, dict) else params:  # noqa: E501
+        for k, v in params.items() if isinstance(params, typing.Mapping) else params:  # noqa: E501
             if k in collection_formats:
                 collection_format = collection_formats[k]
                 if collection_format == 'multi':
@@ -523,7 +523,7 @@ def get_file_data_and_close_file(file_instance: io.IOBase) -> bytes:
         file_instance.close()
         return file_data
 
-    def files_parameters(self, files: typing.Optional[typing.Dict[str, typing.List[io.IOBase]]] = None):
+    def files_parameters(self, files: typing.Optional[typing.Mapping[str, typing.Sequence[io.IOBase]]] = None):
         """Builds form parameters.
 
         :param files: None or a dict with key=param_name and

pinecone/data/index.py

@@ -1,7 +1,7 @@
 from tqdm.autonotebook import tqdm
 
 from collections.abc import Iterable
-from typing import Union, List, Tuple, Optional, Dict, Any
+from typing import Mapping, Union, List, Tuple, Optional, Dict, Any
 
 from pinecone.config import ConfigBuilder
 
@@ -74,12 +74,12 @@ def __init__(
             api_key: str,
             host: str,
             pool_threads: Optional[int] = 1,
-            additional_headers: Optional[Dict[str, str]] = {},
+            additional_headers: Optional[Mapping[str, str]] = {},
             **kwargs
         ):
         self._config = ConfigBuilder.build(api_key=api_key, host=host, **kwargs)
-        
-        api_client = ApiClient(configuration=self._config.openapi_config, 
+
+        api_client = ApiClient(configuration=self._config.openapi_config,
                                pool_threads=pool_threads)
 
         # Configure request headers
@@ -90,7 +90,7 @@ def __init__(
 
         self._api_client = api_client
         self._vector_api = VectorOperationsApi(api_client=api_client)
-    
+
     def __enter__(self):
         return self
 
@@ -245,7 +245,7 @@ def delete(
         ids: Optional[List[str]] = None,
         delete_all: Optional[bool] = None,
         namespace: Optional[str] = None,
-        filter: Optional[Dict[str, Union[str, float, int, bool, List, dict]]] = None,
+        filter: Optional[Mapping[str, Union[str, float, int, bool, List, dict]]] = None,
         **kwargs,
     ) -> Dict[str, Any]:
         """
@@ -272,7 +272,7 @@ def delete(
                                 Default is False.
             namespace (str): The namespace to delete vectors from [optional]
                             If not specified, the default namespace is used.
-            filter (Dict[str, Union[str, float, int, bool, List, dict]]):
+            filter (Mapping[str, Union[str, float, int, bool, List, dict]]):
                     If specified, the metadata filter here will be used to select the vectors to delete.
                     This is mutually exclusive with specifying ids to delete in the ids param or using delete_all=True.
                     See https://www.pinecone.io/docs/metadata-filtering/.. [optional]
@@ -330,10 +330,10 @@ def query(
         vector: Optional[List[float]] = None,
         id: Optional[str] = None,
         namespace: Optional[str] = None,
-        filter: Optional[Dict[str, Union[str, float, int, bool, List, dict]]] = None,
+        filter: Optional[Mapping[str, Union[str, float, int, bool, List, dict]]] = None,
         include_values: Optional[bool] = None,
         include_metadata: Optional[bool] = None,
-        sparse_vector: Optional[Union[SparseValues, Dict[str, Union[List[float], List[int]]]]] = None,
+        sparse_vector: Optional[Union[SparseValues, Mapping[str, Union[List[float], List[int]]]]] = None,
         **kwargs,
     ) -> QueryResponse:
         """
@@ -362,17 +362,17 @@ def query(
             top_k (int): The number of results to return for each query. Must be an integer greater than 1.
             namespace (str): The namespace to fetch vectors from.
                              If not specified, the default namespace is used. [optional]
-            filter (Dict[str, Union[str, float, int, bool, List, dict]):
+            filter (Mapping[str, Union[str, float, int, bool, List, dict]):
                     The filter to apply. You can use vector metadata to limit your search.
                     See https://www.pinecone.io/docs/metadata-filtering/.. [optional]
             include_values (bool): Indicates whether vector values are included in the response.
                                    If omitted the server will use the default value of False [optional]
             include_metadata (bool): Indicates whether metadata is included in the response as well as the ids.
                                      If omitted the server will use the default value of False  [optional]
-            sparse_vector: (Union[SparseValues, Dict[str, Union[List[float], List[int]]]]): sparse values of the query vector.
+            sparse_vector: (Union[SparseValues, Mapping[str, Union[List[float], List[int]]]]): sparse values of the query vector.
                             Expected to be either a SparseValues object or a dict of the form:
                              {'indices': List[int], 'values': List[float]}, where the lists each have the same length.
-        
+
         Returns: QueryResponse object which contains the list of the closest vectors as ScoredVector objects,
                  and namespace name.
         """
@@ -414,9 +414,9 @@ def update(
         self,
         id: str,
         values: Optional[List[float]] = None,
-        set_metadata: Optional[Dict[str, Union[str, float, int, bool, List[int], List[float], List[str]]]] = None,
+        set_metadata: Optional[Mapping[str, Union[str, float, int, bool, List[int], List[float], List[str]]]] = None,
         namespace: Optional[str] = None,
-        sparse_values: Optional[Union[SparseValues, Dict[str, Union[List[float], List[int]]]]] = None,
+        sparse_values: Optional[Union[SparseValues, Mapping[str, Union[List[float], List[int]]]]] = None,
         **kwargs,
     ) -> Dict[str, Any]:
         """
@@ -438,10 +438,10 @@ def update(
         Args:
             id (str): Vector's unique id.
             values (List[float]): vector values to set. [optional]
-            set_metadata (Dict[str, Union[str, float, int, bool, List[int], List[float], List[str]]]]):
+            set_metadata (Mapping[str, Union[str, float, int, bool, List[int], List[float], List[str]]]]):
                 metadata to set for vector. [optional]
             namespace (str): Namespace name where to update the vector.. [optional]
-            sparse_values: (Dict[str, Union[List[float], List[int]]]): sparse values to update for the vector.
+            sparse_values: (Mapping[str, Union[List[float], List[int]]]): sparse values to update for the vector.
                            Expected to be either a SparseValues object or a dict of the form:
                            {'indices': List[int], 'values': List[float]} where the lists each have the same length.
 
@@ -472,7 +472,7 @@ def update(
 
     @validate_and_convert_errors
     def describe_index_stats(
-        self, filter: Optional[Dict[str, Union[str, float, int, bool, List, dict]]] = None, **kwargs
+        self, filter: Optional[Mapping[str, Union[str, float, int, bool, List, dict]]] = None, **kwargs
     ) -> DescribeIndexStatsResponse:
         """
         The DescribeIndexStats operation returns statistics about the index's contents.
@@ -485,7 +485,7 @@ def describe_index_stats(
             >>> index.describe_index_stats(filter={'key': 'value'})
 
         Args:
-            filter (Dict[str, Union[str, float, int, bool, List, dict]]):
+            filter (Mapping[str, Union[str, float, int, bool, List, dict]]):
             If this parameter is present, the operation only returns statistics for vectors that satisfy the filter.
             See https://www.pinecone.io/docs/metadata-filtering/.. [optional]
 
@@ -509,15 +509,15 @@ def _parse_non_empty_args(args: List[Tuple[str, Any]]) -> Dict[str, Any]:
 
     @staticmethod
     def _parse_sparse_values_arg(
-        sparse_values: Optional[Union[SparseValues, Dict[str, Union[List[float], List[int]]]]]
+        sparse_values: Optional[Union[SparseValues, Mapping[str, Union[List[float], List[int]]]]]
     ) -> Optional[SparseValues]:
         if sparse_values is None:
             return None
 
         if isinstance(sparse_values, SparseValues):
             return sparse_values
 
-        if not isinstance(sparse_values, dict) or "indices" not in sparse_values or "values" not in sparse_values:
+        if not isinstance(sparse_values, Mapping) or "indices" not in sparse_values or "values" not in sparse_values:
             raise ValueError(
                 "Invalid sparse values argument. Expected a dict of: {'indices': List[int], 'values': List[float]}."
                 f"Received: {sparse_values}"

pinecone/grpc/index_grpc.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Optional, Dict, Union, List, Tuple, Any, TypedDict, cast
+from typing import Mapping, Optional, Dict, Union, List, Tuple, Any, TypedDict, cast
 
 from google.protobuf import json_format
 
@@ -208,7 +208,7 @@ def delete(
         ids: Optional[List[str]] = None,
         delete_all: Optional[bool] = None,
         namespace: Optional[str] = None,
-        filter: Optional[Dict[str, Union[str, float, int, bool, List, dict]]] = None,
+        filter: Optional[Mapping[str, Union[str, float, int, bool, List, dict]]] = None,
         async_req: bool = False,
         **kwargs,
     ) -> Union[DeleteResponse, PineconeGrpcFuture]:
@@ -234,7 +234,7 @@ def delete(
                                Default is False.
             namespace (str): The namespace to delete vectors from [optional]
                              If not specified, the default namespace is used.
-            filter (Dict[str, Union[str, float, int, bool, List, dict]]):
+            filter (Mapping[str, Union[str, float, int, bool, List, dict]]):
                     If specified, the metadata filter here will be used to select the vectors to delete.
                     This is mutually exclusive with specifying ids to delete in the ids param or using delete_all=True.
                      See https://www.pinecone.io/docs/metadata-filtering/.. [optional]
@@ -292,7 +292,7 @@ def query(
         id: Optional[str] = None,
         namespace: Optional[str] = None,
         top_k: Optional[int] = None,
-        filter: Optional[Dict[str, Union[str, float, int, bool, List, dict]]] = None,
+        filter: Optional[Mapping[str, Union[str, float, int, bool, List, dict]]] = None,
         include_values: Optional[bool] = None,
         include_metadata: Optional[bool] = None,
         sparse_vector: Optional[Union[GRPCSparseValues, SparseVectorTypedDict]] = None,
@@ -322,7 +322,7 @@ def query(
             top_k (int): The number of results to return for each query. Must be an integer greater than 1.
             namespace (str): The namespace to fetch vectors from.
                              If not specified, the default namespace is used. [optional]
-            filter (Dict[str, Union[str, float, int, bool, List, dict]]):
+            filter (Mapping[str, Union[str, float, int, bool, List, dict]]):
                     The filter to apply. You can use vector metadata to limit your search.
                     See https://www.pinecone.io/docs/metadata-filtering/.. [optional]
             include_values (bool): Indicates whether vector values are included in the response.
@@ -371,7 +371,7 @@ def update(
         id: str,
         async_req: bool = False,
         values: Optional[List[float]] = None,
-        set_metadata: Optional[Dict[str, Union[str, float, int, bool, List[int], List[float], List[str]]]] = None,
+        set_metadata: Optional[Mapping[str, Union[str, float, int, bool, List[int], List[float], List[str]]]] = None,
         namespace: Optional[str] = None,
         sparse_values: Optional[Union[GRPCSparseValues, SparseVectorTypedDict]] = None,
         **kwargs,
@@ -395,7 +395,7 @@ def update(
             async_req (bool): If True, the update operation will be performed asynchronously.
                               Defaults to False. [optional]
             values (List[float]): vector values to set. [optional]
-            set_metadata (Dict[str, Union[str, float, int, bool, List[int], List[float], List[str]]]]):
+            set_metadata (Mapping[str, Union[str, float, int, bool, List[int], List[float], List[str]]]]):
                 metadata to set for vector. [optional]
             namespace (str): Namespace name where to update the vector.. [optional]
             sparse_values: (Dict[str, Union[List[float], List[int]]]): sparse values to update for the vector.
@@ -429,7 +429,7 @@ def update(
             return self._wrap_grpc_call(self.stub.Update, request, timeout=timeout)
 
     def describe_index_stats(
-        self, filter: Optional[Dict[str, Union[str, float, int, bool, List, dict]]] = None, **kwargs
+        self, filter: Optional[Mapping[str, Union[str, float, int, bool, List, dict]]] = None, **kwargs
     ) -> DescribeIndexStatsResponse:
         """
         The DescribeIndexStats operation returns statistics about the index's contents.
@@ -440,7 +440,7 @@ def describe_index_stats(
             >>> index.describe_index_stats(filter={'key': 'value'})
 
         Args:
-            filter (Dict[str, Union[str, float, int, bool, List, dict]]):
+            filter (Mapping[str, Union[str, float, int, bool, List, dict]]):
             If this parameter is present, the operation only returns statistics for vectors that satisfy the filter.
             See https://www.pinecone.io/docs/metadata-filtering/.. [optional]