pinecone.core.grpc.index_grpc

  1import logging
  2import numbers
  3from abc import ABC, abstractmethod
  4from functools import wraps
  5from importlib.util import find_spec
  6from typing import NamedTuple, Optional, Dict, Iterable, Union, List, Tuple, Any
  7from collections.abc import Mapping
  8
  9import certifi
 10import grpc
 11from google.protobuf import json_format
 12from grpc._channel import _InactiveRpcError, _MultiThreadedRendezvous
 13from tqdm.autonotebook import tqdm
 14import json
 15
 16from pinecone import FetchResponse, QueryResponse, ScoredVector, SingleQueryResults, DescribeIndexStatsResponse
 17from pinecone.config import Config
 18from pinecone.core.client.model.namespace_summary import NamespaceSummary
 19from pinecone.core.client.model.vector import Vector as _Vector
 20from pinecone.core.grpc.protos.vector_service_pb2 import (
 21    Vector as GRPCVector,
 22    QueryVector as GRPCQueryVector,
 23    UpsertRequest,
 24    UpsertResponse,
 25    DeleteRequest,
 26    QueryRequest,
 27    FetchRequest,
 28    UpdateRequest,
 29    DescribeIndexStatsRequest,
 30    DeleteResponse,
 31    UpdateResponse,
 32    SparseValues as GRPCSparseValues,
 33)
 34from pinecone.core.client.model.sparse_values import SparseValues
 35from pinecone.core.grpc.protos.vector_service_pb2_grpc import VectorServiceStub
 36from pinecone.core.grpc.retry import RetryOnRpcErrorClientInterceptor, RetryConfig
 37from pinecone.core.utils import _generate_request_id, dict_to_proto_struct, fix_tuple_length
 38from pinecone.core.utils.constants import (
 39    MAX_MSG_SIZE,
 40    REQUEST_ID,
 41    CLIENT_VERSION,
 42    REQUIRED_VECTOR_FIELDS,
 43    OPTIONAL_VECTOR_FIELDS,
 44)
 45from pinecone.exceptions import PineconeException
 46
 47__all__ = ["GRPCIndex", "GRPCVector", "GRPCQueryVector", "GRPCSparseValues"]
 48
 49_logger = logging.getLogger(__name__)
 50
 51
 52class GRPCClientConfig(NamedTuple):
 53    """
 54    GRPC client configuration options.
 55
 56    :param secure: Whether to use encrypted protocol (SSL). defaults to True.
 57    :type traceroute: bool, optional
 58    :param timeout: defaults to 2 seconds. Fail if gateway doesn't receive response within timeout.
 59    :type timeout: int, optional
 60    :param conn_timeout: defaults to 1. Timeout to retry connection if gRPC is unavailable. 0 is no retry.
 61    :type conn_timeout: int, optional
 62    :param reuse_channel: Whether to reuse the same grpc channel for multiple requests
 63    :type reuse_channel: bool, optional
 64    :param retry_config: RetryConfig indicating how requests should be retried
 65    :type retry_config: RetryConfig, optional
 66    :param grpc_channel_options: A dict of gRPC channel arguments
 67    :type grpc_channel_options: Dict[str, str]
 68    """
 69
 70    secure: bool = True
 71    timeout: int = 20
 72    conn_timeout: int = 1
 73    reuse_channel: bool = True
 74    retry_config: Optional[RetryConfig] = None
 75    grpc_channel_options: Dict[str, str] = None
 76
 77    @classmethod
 78    def _from_dict(cls, kwargs: dict):
 79        cls_kwargs = {kk: vv for kk, vv in kwargs.items() if kk in cls._fields}
 80        return cls(**cls_kwargs)
 81
 82
 83class GRPCIndexBase(ABC):
 84    """
 85    Base class for grpc-based interaction with Pinecone indexes
 86    """
 87
 88    _pool = None
 89
 90    def __init__(
 91        self, index_name: str, channel=None, grpc_config: GRPCClientConfig = None, _endpoint_override: str = None
 92    ):
 93        self.name = index_name
 94
 95        self.grpc_client_config = grpc_config or GRPCClientConfig()
 96        self.retry_config = self.grpc_client_config.retry_config or RetryConfig()
 97        self.fixed_metadata = {"api-key": Config.API_KEY, "service-name": index_name, "client-version": CLIENT_VERSION}
 98        self._endpoint_override = _endpoint_override
 99
100        self.method_config = json.dumps(
101            {
102                "methodConfig": [
103                    {
104                        "name": [{"service": "VectorService.Upsert"}],
105                        "retryPolicy": {
106                            "maxAttempts": 5,
107                            "initialBackoff": "0.1s",
108                            "maxBackoff": "1s",
109                            "backoffMultiplier": 2,
110                            "retryableStatusCodes": ["UNAVAILABLE"],
111                        },
112                    },
113                    {
114                        "name": [{"service": "VectorService"}],
115                        "retryPolicy": {
116                            "maxAttempts": 5,
117                            "initialBackoff": "0.1s",
118                            "maxBackoff": "1s",
119                            "backoffMultiplier": 2,
120                            "retryableStatusCodes": ["UNAVAILABLE"],
121                        },
122                    },
123                ]
124            }
125        )
126
127        self._channel = channel or self._gen_channel()
128        self.stub = self.stub_class(self._channel)
129
130    @property
131    @abstractmethod
132    def stub_class(self):
133        pass
134
135    def _endpoint(self):
136        return (
137            self._endpoint_override
138            if self._endpoint_override
139            else f"{self.name}-{Config.PROJECT_NAME}.svc.{Config.ENVIRONMENT}.pinecone.io:443"
140        )
141
142    def _gen_channel(self, options=None):
143        target = self._endpoint()
144        default_options = {
145            "grpc.max_send_message_length": MAX_MSG_SIZE,
146            "grpc.max_receive_message_length": MAX_MSG_SIZE,
147            "grpc.service_config": self.method_config,
148            "grpc.enable_retries": True,
149        }
150        if self.grpc_client_config.secure:
151            default_options["grpc.ssl_target_name_override"] = target.split(":")[0]
152        user_provided_options = options or {}
153        _options = tuple((k, v) for k, v in {**default_options, **user_provided_options}.items())
154        _logger.debug(
155            "creating new channel with endpoint %s options %s and config %s", target, _options, self.grpc_client_config
156        )
157        if not self.grpc_client_config.secure:
158            channel = grpc.insecure_channel(target, options=_options)
159        else:
160            root_cas = open(certifi.where(), "rb").read()
161            tls = grpc.ssl_channel_credentials(root_certificates=root_cas)
162            channel = grpc.secure_channel(target, tls, options=_options)
163
164        return channel
165
166    @property
167    def channel(self):
168        """Creates GRPC channel."""
169        if self.grpc_client_config.reuse_channel and self._channel and self.grpc_server_on():
170            return self._channel
171        self._channel = self._gen_channel()
172        return self._channel
173
174    def grpc_server_on(self) -> bool:
175        try:
176            grpc.channel_ready_future(self._channel).result(timeout=self.grpc_client_config.conn_timeout)
177            return True
178        except grpc.FutureTimeoutError:
179            return False
180
181    def close(self):
182        """Closes the connection to the index."""
183        try:
184            self._channel.close()
185        except TypeError:
186            pass
187
188    def _wrap_grpc_call(
189        self, func, request, timeout=None, metadata=None, credentials=None, wait_for_ready=None, compression=None
190    ):
191        @wraps(func)
192        def wrapped():
193            user_provided_metadata = metadata or {}
194            _metadata = tuple(
195                (k, v) for k, v in {**self.fixed_metadata, **self._request_metadata(), **user_provided_metadata}.items()
196            )
197            try:
198                return func(
199                    request,
200                    timeout=timeout,
201                    metadata=_metadata,
202                    credentials=credentials,
203                    wait_for_ready=wait_for_ready,
204                    compression=compression,
205                )
206            except _InactiveRpcError as e:
207                raise PineconeException(e._state.debug_error_string) from e
208
209        return wrapped()
210
211    def _request_metadata(self) -> Dict[str, str]:
212        return {REQUEST_ID: _generate_request_id()}
213
214    def __enter__(self):
215        return self
216
217    def __exit__(self, exc_type, exc_value, traceback):
218        self.close()
219
220
221def parse_sparse_values(sparse_values: dict):
222    return (
223        SparseValues(indices=sparse_values["indices"], values=sparse_values["values"])
224        if sparse_values
225        else SparseValues(indices=[], values=[])
226    )
227
228
229def parse_fetch_response(response: dict):
230    vd = {}
231    vectors = response.get("vectors")
232    if not vectors:
233        return None
234    for id, vec in vectors.items():
235        v_obj = _Vector(
236            id=vec["id"],
237            values=vec["values"],
238            sparse_values=parse_sparse_values(vec.get("sparseValues")),
239            metadata=vec.get("metadata", None),
240            _check_type=False,
241        )
242        vd[id] = v_obj
243    namespace = response.get("namespace", "")
244    return FetchResponse(vectors=vd, namespace=namespace, _check_type=False)
245
246
247def parse_query_response(response: dict, unary_query: bool, _check_type: bool = False):
248    res = []
249
250    # TODO: consider deleting this deprecated case
251    for match in response.get("results", []):
252        namespace = match.get("namespace", "")
253        m = []
254        if "matches" in match:
255            for item in match["matches"]:
256                sc = ScoredVector(
257                    id=item["id"],
258                    score=item.get("score", 0.0),
259                    values=item.get("values", []),
260                    sparse_values=parse_sparse_values(item.get("sparseValues")),
261                    metadata=item.get("metadata", {}),
262                )
263                m.append(sc)
264        res.append(SingleQueryResults(matches=m, namespace=namespace))
265
266    m = []
267    for item in response.get("matches", []):
268        sc = ScoredVector(
269            id=item["id"],
270            score=item.get("score", 0.0),
271            values=item.get("values", []),
272            sparse_values=parse_sparse_values(item.get("sparseValues")),
273            metadata=item.get("metadata", {}),
274            _check_type=_check_type,
275        )
276        m.append(sc)
277
278    kwargs = {"_check_type": _check_type}
279    if unary_query:
280        kwargs["namespace"] = response.get("namespace", "")
281        kwargs["matches"] = m
282    else:
283        kwargs["results"] = res
284    return QueryResponse(**kwargs)
285
286
287def parse_stats_response(response: dict):
288    fullness = response.get("indexFullness", 0.0)
289    total_vector_count = response.get("totalVectorCount", 0)
290    dimension = response.get("dimension", 0)
291    summaries = response.get("namespaces", {})
292    namespace_summaries = {}
293    for key in summaries:
294        vc = summaries[key].get("vectorCount", 0)
295        namespace_summaries[key] = NamespaceSummary(vector_count=vc)
296    return DescribeIndexStatsResponse(
297        namespaces=namespace_summaries,
298        dimension=dimension,
299        index_fullness=fullness,
300        total_vector_count=total_vector_count,
301        _check_type=False,
302    )
303
304
305class PineconeGrpcFuture:
306    def __init__(self, delegate):
307        self._delegate = delegate
308
309    def cancel(self):
310        return self._delegate.cancel()
311
312    def cancelled(self):
313        return self._delegate.cancelled()
314
315    def running(self):
316        return self._delegate.running()
317
318    def done(self):
319        return self._delegate.done()
320
321    def add_done_callback(self, fun):
322        return self._delegate.add_done_callback(fun)
323
324    def result(self, timeout=None):
325        try:
326            return self._delegate.result(timeout=timeout)
327        except _MultiThreadedRendezvous as e:
328            raise PineconeException(e._state.debug_error_string) from e
329
330    def exception(self, timeout=None):
331        return self._delegate.exception(timeout=timeout)
332
333    def traceback(self, timeout=None):
334        return self._delegate.traceback(timeout=timeout)
335
336
337class GRPCIndex(GRPCIndexBase):
338
339    """A client for interacting with a Pinecone index via GRPC API."""
340
341    @property
342    def stub_class(self):
343        return VectorServiceStub
344
345    def upsert(
346        self,
347        vectors: Union[List[GRPCVector], List[tuple], List[dict]],
348        async_req: bool = False,
349        namespace: Optional[str] = None,
350        batch_size: Optional[int] = None,
351        show_progress: bool = True,
352        **kwargs
353    ) -> Union[UpsertResponse, PineconeGrpcFuture]:
354        """
355        The upsert operation writes vectors into a namespace.
356        If a new value is upserted for an existing vector id, it will overwrite the previous value.
357
358        Examples:
359            >>> index.upsert([('id1', [1.0, 2.0, 3.0], {'key': 'value'}),
360                              ('id2', [1.0, 2.0, 3.0])
361                              ],
362                              namespace='ns1', async_req=True)
363            >>> index.upsert([{'id': 'id1', 'values': [1.0, 2.0, 3.0], 'metadata': {'key': 'value'}},
364                              {'id': 'id2',
365                                        'values': [1.0, 2.0, 3.0],
366                                        'sprase_values': {'indices': [1, 8], 'values': [0.2, 0.4]},
367                              ])
368            >>> index.upsert([GRPCVector(id='id1', values=[1.0, 2.0, 3.0], metadata={'key': 'value'}),
369                              GRPCVector(id='id2', values=[1.0, 2.0, 3.0]),
370                              GRPCVector(id='id3',
371                                         values=[1.0, 2.0, 3.0],
372                                         sparse_values=GRPCSparseValues(indices=[1, 2], values=[0.2, 0.4]))])
373
374        Args:
375            vectors (Union[List[Vector], List[Tuple]]): A list of vectors to upsert.
376
377                     A vector can be represented by a 1) GRPCVector object, a 2) tuple or 3) a dictionary
378                     1) if a tuple is used, it must be of the form (id, values, metadata) or (id, values).
379                        where id is a string, vector is a list of floats, and metadata is a dict.
380                        Examples: ('id1', [1.0, 2.0, 3.0], {'key': 'value'}), ('id2', [1.0, 2.0, 3.0])
381
382                    2) if a GRPCVector object is used, a GRPCVector object must be of the form
383                        GRPCVector(id, values, metadata), where metadata is an optional argument of type
384                        Dict[str, Union[str, float, int, bool, List[int], List[float], List[str]]]
385                       Examples: GRPCVector(id='id1', values=[1.0, 2.0, 3.0], metadata={'key': 'value'}),
386                                 GRPCVector(id='id2', values=[1.0, 2.0, 3.0]),
387                                 GRPCVector(id='id3',
388                                            values=[1.0, 2.0, 3.0],
389                                            sparse_values=GRPCSparseValues(indices=[1, 2], values=[0.2, 0.4]))
390
391                    3) if a dictionary is used, it must be in the form
392                       {'id': str, 'values': List[float], 'sparse_values': {'indices': List[int], 'values': List[float]},
393                        'metadata': dict}
394
395                    Note: the dimension of each vector must match the dimension of the index.
396            async_req (bool): If True, the upsert operation will be performed asynchronously.
397                              Cannot be used with batch_size.
398                              Defaults to False. See: https://docs.pinecone.io/docs/performance-tuning [optional]
399            namespace (str): The namespace to write to. If not specified, the default namespace is used. [optional]
400            batch_size (int): The number of vectors to upsert in each batch.
401                                Cannot be used with async_req=Ture.
402                               If not specified, all vectors will be upserted in a single batch. [optional]
403            show_progress (bool): Whether to show a progress bar using tqdm.
404                                  Applied only if batch_size is provided. Default is True.
405
406        Returns: UpsertResponse, contains the number of vectors upserted
407        """
408        if async_req and batch_size is not None:
409            raise ValueError(
410                "async_req is not supported when batch_size is provided."
411                "To upsert in parallel, please follow: "
412                "https://docs.pinecone.io/docs/performance-tuning"
413            )
414
415        def _dict_to_grpc_vector(item):
416            item_keys = set(item.keys())
417            if not item_keys.issuperset(REQUIRED_VECTOR_FIELDS):
418                raise ValueError(
419                    f"Vector dictionary is missing required fields: {list(REQUIRED_VECTOR_FIELDS - item_keys)}"
420                )
421
422            excessive_keys = item_keys - (REQUIRED_VECTOR_FIELDS | OPTIONAL_VECTOR_FIELDS)
423            if len(excessive_keys) > 0:
424                raise ValueError(
425                    f"Found excess keys in the vector dictionary: {list(excessive_keys)}. "
426                    f"The allowed keys are: {list(REQUIRED_VECTOR_FIELDS | OPTIONAL_VECTOR_FIELDS)}"
427                )
428
429            sparse_values = None
430            if "sparse_values" in item:
431                if not isinstance(item["sparse_values"], Mapping):
432                    raise TypeError(
433                        f"Column `sparse_values` is expected to be a dictionary, found {type(item['sparse_values'])}"
434                    )
435                indices = item["sparse_values"].get("indices", None)
436                values = item["sparse_values"].get("values", None)
437                try:
438                    sparse_values = GRPCSparseValues(indices=indices, values=values)
439                except TypeError as e:
440                    raise TypeError(
441                        "Found unexpected data in column `sparse_values`. "
442                        "Expected format is `'sparse_values': {'indices': List[int], 'values': List[float]}`."
443                    ) from e
444
445            metadata = item.get("metadata", None)
446            if metadata is not None and not isinstance(metadata, Mapping):
447                raise TypeError(f"Column `metadata` is expected to be a dictionary, found {type(metadata)}")
448
449            try:
450                return GRPCVector(
451                    id=item["id"],
452                    values=item["values"],
453                    sparse_values=sparse_values,
454                    metadata=dict_to_proto_struct(metadata),
455                )
456
457            except TypeError as e:
458                # No need to raise a dedicated error for `id` - protobuf's error message is clear enough
459                if not isinstance(item["values"], Iterable) or not isinstance(item["values"][0], numbers.Real):
460                    raise TypeError(f"Column `values` is expected to be a list of floats")
461                raise
462
463        def _vector_transform(item):
464            if isinstance(item, GRPCVector):
465                return item
466            elif isinstance(item, tuple):
467                if len(item) > 3:
468                    raise ValueError(
469                        f"Found a tuple of length {len(item)} which is not supported. "
470                        f"Vectors can be represented as tuples either the form (id, values, metadata) or (id, values). "
471                        f"To pass sparse values please use either dicts or GRPCVector objects as inputs."
472                    )
473                id, values, metadata = fix_tuple_length(item, 3)
474                return GRPCVector(id=id, values=values, metadata=dict_to_proto_struct(metadata) or {})
475            elif isinstance(item, Mapping):
476                return _dict_to_grpc_vector(item)
477            raise ValueError(f"Invalid vector value passed: cannot interpret type {type(item)}")
478
479        timeout = kwargs.pop("timeout", None)
480
481        vectors = list(map(_vector_transform, vectors))
482        if async_req:
483            args_dict = self._parse_non_empty_args([("namespace", namespace)])
484            request = UpsertRequest(vectors=vectors, **args_dict, **kwargs)
485            future = self._wrap_grpc_call(self.stub.Upsert.future, request, timeout=timeout)
486            return PineconeGrpcFuture(future)
487
488        if batch_size is None:
489            return self._upsert_batch(vectors, namespace, timeout=timeout, **kwargs)
490
491        if not isinstance(batch_size, int) or batch_size <= 0:
492            raise ValueError("batch_size must be a positive integer")
493
494        pbar = tqdm(total=len(vectors), disable=not show_progress, desc="Upserted vectors")
495        total_upserted = 0
496        for i in range(0, len(vectors), batch_size):
497            batch_result = self._upsert_batch(vectors[i : i + batch_size], namespace, timeout=timeout, **kwargs)
498            pbar.update(batch_result.upserted_count)
499            # we can't use here pbar.n for the case show_progress=False
500            total_upserted += batch_result.upserted_count
501
502        return UpsertResponse(upserted_count=total_upserted)
503
504    def _upsert_batch(
505        self, vectors: List[GRPCVector], namespace: Optional[str], timeout: Optional[float], **kwargs
506    ) -> UpsertResponse:
507        args_dict = self._parse_non_empty_args([("namespace", namespace)])
508        request = UpsertRequest(vectors=vectors, **args_dict)
509        return self._wrap_grpc_call(self.stub.Upsert, request, timeout=timeout, **kwargs)
510
511    def upsert_from_dataframe(
512        self,
513        df,
514        namespace: str = None,
515        batch_size: int = 500,
516        use_async_requests: bool = True,
517        show_progress: bool = True,
518    ) -> None:
519        """Upserts a dataframe into the index.
520
521        Args:
522            df: A pandas dataframe with the following columns: id, vector, and metadata.
523            namespace: The namespace to upsert into.
524            batch_size: The number of rows to upsert in a single batch.
525            use_async_requests: Whether to upsert multiple requests at the same time using asynchronous request mechanism.
526                                Set to `False`
527            show_progress: Whether to show a progress bar.
528        """
529        try:
530            import pandas as pd
531        except ImportError:
532            raise RuntimeError(
533                "The `pandas` package is not installed. Please install pandas to use `upsert_from_dataframe()`"
534            )
535
536        if not isinstance(df, pd.DataFrame):
537            raise ValueError(f"Only pandas dataframes are supported. Found: {type(df)}")
538
539        pbar = tqdm(total=len(df), disable=not show_progress, desc="sending upsert requests")
540        results = []
541        for chunk in self._iter_dataframe(df, batch_size=batch_size):
542            res = self.upsert(vectors=chunk, namespace=namespace, async_req=use_async_requests)
543            pbar.update(len(chunk))
544            results.append(res)
545
546        if use_async_requests:
547            results = [async_result.result() for async_result in tqdm(results, desc="collecting async responses")]
548
549        upserted_count = 0
550        for res in results:
551            upserted_count += res.upserted_count
552
553        return UpsertResponse(upserted_count=upserted_count)
554
555    @staticmethod
556    def _iter_dataframe(df, batch_size):
557        for i in range(0, len(df), batch_size):
558            batch = df.iloc[i : i + batch_size].to_dict(orient="records")
559            yield batch
560
561    def delete(
562        self,
563        ids: Optional[List[str]] = None,
564        delete_all: Optional[bool] = None,
565        namespace: Optional[str] = None,
566        filter: Optional[Dict[str, Union[str, float, int, bool, List, dict]]] = None,
567        async_req: bool = False,
568        **kwargs
569    ) -> Union[DeleteResponse, PineconeGrpcFuture]:
570        """
571        The Delete operation deletes vectors from the index, from a single namespace.
572        No error raised if the vector id does not exist.
573        Note: for any delete call, if namespace is not specified, the default namespace is used.
574
575        Delete can occur in the following mutual exclusive ways:
576        1. Delete by ids from a single namespace
577        2. Delete all vectors from a single namespace by setting delete_all to True
578        3. Delete all vectors from a single namespace by specifying a metadata filter
579           (note that for this option delete all must be set to False)
580
581        Examples:
582            >>> index.delete(ids=['id1', 'id2'], namespace='my_namespace')
583            >>> index.delete(delete_all=True, namespace='my_namespace')
584            >>> index.delete(filter={'key': 'value'}, namespace='my_namespace', async_req=True)
585
586        Args:
587            ids (List[str]): Vector ids to delete [optional]
588            delete_all (bool): This indicates that all vectors in the index namespace should be deleted.. [optional]
589                               Default is False.
590            namespace (str): The namespace to delete vectors from [optional]
591                             If not specified, the default namespace is used.
592            filter (Dict[str, Union[str, float, int, bool, List, dict]]):
593                    If specified, the metadata filter here will be used to select the vectors to delete.
594                    This is mutually exclusive with specifying ids to delete in the ids param or using delete_all=True.
595                     See https://www.pinecone.io/docs/metadata-filtering/.. [optional]
596            async_req (bool): If True, the delete operation will be performed asynchronously.
597                              Defaults to False. [optional]
598
599        Returns: DeleteResponse (contains no data) or a PineconeGrpcFuture object if async_req is True.
600        """
601
602        if filter is not None:
603            filter = dict_to_proto_struct(filter)
604
605        args_dict = self._parse_non_empty_args(
606            [("ids", ids), ("delete_all", delete_all), ("namespace", namespace), ("filter", filter)]
607        )
608        timeout = kwargs.pop("timeout", None)
609
610        request = DeleteRequest(**args_dict, **kwargs)
611        if async_req:
612            future = self._wrap_grpc_call(self.stub.Delete.future, request, timeout=timeout)
613            return PineconeGrpcFuture(future)
614        else:
615            return self._wrap_grpc_call(self.stub.Delete, request, timeout=timeout)
616
617    def fetch(self, ids: Optional[List[str]], namespace: Optional[str] = None, **kwargs) -> FetchResponse:
618        """
619        The fetch operation looks up and returns vectors, by ID, from a single namespace.
620        The returned vectors include the vector data and/or metadata.
621
622        Examples:
623            >>> index.fetch(ids=['id1', 'id2'], namespace='my_namespace')
624            >>> index.fetch(ids=['id1', 'id2'])
625
626        Args:
627            ids (List[str]): The vector IDs to fetch.
628            namespace (str): The namespace to fetch vectors from.
629                             If not specified, the default namespace is used. [optional]
630
631        Returns: FetchResponse object which contains the list of Vector objects, and namespace name.
632        """
633        timeout = kwargs.pop("timeout", None)
634
635        args_dict = self._parse_non_empty_args([("namespace", namespace)])
636
637        request = FetchRequest(ids=ids, **args_dict, **kwargs)
638        response = self._wrap_grpc_call(self.stub.Fetch, request, timeout=timeout)
639        json_response = json_format.MessageToDict(response)
640        return parse_fetch_response(json_response)
641
642    def query(
643        self,
644        vector: Optional[List[float]] = None,
645        id: Optional[str] = None,
646        queries: Optional[Union[List[GRPCQueryVector], List[Tuple]]] = None,
647        namespace: Optional[str] = None,
648        top_k: Optional[int] = None,
649        filter: Optional[Dict[str, Union[str, float, int, bool, List, dict]]] = None,
650        include_values: Optional[bool] = None,
651        include_metadata: Optional[bool] = None,
652        sparse_vector: Optional[Union[GRPCSparseValues, Dict[str, Union[List[float], List[int]]]]] = None,
653        **kwargs
654    ) -> QueryResponse:
655        """
656        The Query operation searches a namespace, using a query vector.
657        It retrieves the ids of the most similar items in a namespace, along with their similarity scores.
658
659        Examples:
660            >>> index.query(vector=[1, 2, 3], top_k=10, namespace='my_namespace')
661            >>> index.query(id='id1', top_k=10, namespace='my_namespace')
662            >>> index.query(vector=[1, 2, 3], top_k=10, namespace='my_namespace', filter={'key': 'value'})
663            >>> index.query(id='id1', top_k=10, namespace='my_namespace', include_metadata=True, include_values=True)
664            >>> index.query(vector=[1, 2, 3], sparse_vector={'indices': [1, 2], 'values': [0.2, 0.4]},
665            >>>             top_k=10, namespace='my_namespace')
666            >>> index.query(vector=[1, 2, 3], sparse_vector=GRPCSparseValues([1, 2], [0.2, 0.4]),
667            >>>             top_k=10, namespace='my_namespace')
668
669        Args:
670            vector (List[float]): The query vector. This should be the same length as the dimension of the index
671                                  being queried. Each `query()` request can contain only one of the parameters
672                                  `queries`, `id` or `vector`.. [optional]
673            id (str): The unique ID of the vector to be used as a query vector.
674                      Each `query()` request can contain only one of the parameters
675                      `queries`, `vector`, or  `id`.. [optional]
676            queries ([GRPCQueryVector]): DEPRECATED. The query vectors.
677                                     Each `query()` request can contain only one of the parameters
678                                     `queries`, `vector`, or  `id`.. [optional]
679            top_k (int): The number of results to return for each query. Must be an integer greater than 1.
680            namespace (str): The namespace to fetch vectors from.
681                             If not specified, the default namespace is used. [optional]
682            filter (Dict[str, Union[str, float, int, bool, List, dict]]):
683                    The filter to apply. You can use vector metadata to limit your search.
684                    See https://www.pinecone.io/docs/metadata-filtering/.. [optional]
685            include_values (bool): Indicates whether vector values are included in the response.
686                                   If omitted the server will use the default value of False [optional]
687            include_metadata (bool): Indicates whether metadata is included in the response as well as the ids.
688                                     If omitted the server will use the default value of False  [optional]
689            sparse_vector: (Union[SparseValues, Dict[str, Union[List[float], List[int]]]]): sparse values of the query vector.
690                            Expected to be either a GRPCSparseValues object or a dict of the form:
691                             {'indices': List[int], 'values': List[float]}, where the lists each have the same length.
692
693        Returns: QueryResponse object which contains the list of the closest vectors as ScoredVector objects,
694                 and namespace name.
695        """
696
697        def _query_transform(item):
698            if isinstance(item, GRPCQueryVector):
699                return item
700            if isinstance(item, tuple):
701                values, filter = fix_tuple_length(item, 2)
702                filter = dict_to_proto_struct(filter)
703                return GRPCQueryVector(values=values, filter=filter)
704            if isinstance(item, Iterable):
705                return GRPCQueryVector(values=item)
706            raise ValueError(f"Invalid query vector value passed: cannot interpret type {type(item)}")
707
708        queries = list(map(_query_transform, queries)) if queries is not None else None
709
710        if filter is not None:
711            filter = dict_to_proto_struct(filter)
712
713        sparse_vector = self._parse_sparse_values_arg(sparse_vector)
714        args_dict = self._parse_non_empty_args(
715            [
716                ("vector", vector),
717                ("id", id),
718                ("queries", queries),
719                ("namespace", namespace),
720                ("top_k", top_k),
721                ("filter", filter),
722                ("include_values", include_values),
723                ("include_metadata", include_metadata),
724                ("sparse_vector", sparse_vector),
725            ]
726        )
727
728        request = QueryRequest(**args_dict)
729
730        timeout = kwargs.pop("timeout", None)
731        response = self._wrap_grpc_call(self.stub.Query, request, timeout=timeout)
732        json_response = json_format.MessageToDict(response)
733        return parse_query_response(json_response, vector is not None or id, _check_type=False)
734
735    def update(
736        self,
737        id: str,
738        async_req: bool = False,
739        values: Optional[List[float]] = None,
740        set_metadata: Optional[Dict[str, Union[str, float, int, bool, List[int], List[float], List[str]]]] = None,
741        namespace: Optional[str] = None,
742        sparse_values: Optional[Union[GRPCSparseValues, Dict[str, Union[List[float], List[int]]]]] = None,
743        **kwargs
744    ) -> Union[UpdateResponse, PineconeGrpcFuture]:
745        """
746        The Update operation updates vector in a namespace.
747        If a value is included, it will overwrite the previous value.
748        If a set_metadata is included,
749        the values of the fields specified in it will be added or overwrite the previous value.
750
751        Examples:
752            >>> index.update(id='id1', values=[1, 2, 3], namespace='my_namespace')
753            >>> index.update(id='id1', set_metadata={'key': 'value'}, namespace='my_namespace', async_req=True)
754            >>> index.update(id='id1', values=[1, 2, 3], sparse_values={'indices': [1, 2], 'values': [0.2, 0.4]},
755            >>>              namespace='my_namespace')
756            >>> index.update(id='id1', values=[1, 2, 3], sparse_values=GRPCSparseValues(indices=[1, 2], values=[0.2, 0.4]),
757            >>>              namespace='my_namespace')
758
759        Args:
760            id (str): Vector's unique id.
761            async_req (bool): If True, the update operation will be performed asynchronously.
762                              Defaults to False. [optional]
763            values (List[float]): vector values to set. [optional]
764            set_metadata (Dict[str, Union[str, float, int, bool, List[int], List[float], List[str]]]]):
765                metadata to set for vector. [optional]
766            namespace (str): Namespace name where to update the vector.. [optional]
767            sparse_values: (Dict[str, Union[List[float], List[int]]]): sparse values to update for the vector.
768                           Expected to be either a GRPCSparseValues object or a dict of the form:
769                           {'indices': List[int], 'values': List[float]} where the lists each have the same length.
770
771
772        Returns: UpdateResponse (contains no data) or a PineconeGrpcFuture object if async_req is True.
773        """
774        if set_metadata is not None:
775            set_metadata = dict_to_proto_struct(set_metadata)
776        timeout = kwargs.pop("timeout", None)
777
778        sparse_values = self._parse_sparse_values_arg(sparse_values)
779        args_dict = self._parse_non_empty_args(
780            [
781                ("values", values),
782                ("set_metadata", set_metadata),
783                ("namespace", namespace),
784                ("sparse_values", sparse_values),
785            ]
786        )
787
788        request = UpdateRequest(id=id, **args_dict)
789        if async_req:
790            future = self._wrap_grpc_call(self.stub.Update.future, request, timeout=timeout)
791            return PineconeGrpcFuture(future)
792        else:
793            return self._wrap_grpc_call(self.stub.Update, request, timeout=timeout)
794
795    def describe_index_stats(
796        self, filter: Optional[Dict[str, Union[str, float, int, bool, List, dict]]] = None, **kwargs
797    ) -> DescribeIndexStatsResponse:
798        """
799        The DescribeIndexStats operation returns statistics about the index's contents.
800        For example: The vector count per namespace and the number of dimensions.
801
802        Examples:
803            >>> index.describe_index_stats()
804            >>> index.describe_index_stats(filter={'key': 'value'})
805
806        Args:
807            filter (Dict[str, Union[str, float, int, bool, List, dict]]):
808            If this parameter is present, the operation only returns statistics for vectors that satisfy the filter.
809            See https://www.pinecone.io/docs/metadata-filtering/.. [optional]
810
811        Returns: DescribeIndexStatsResponse object which contains stats about the index.
812        """
813        if filter is not None:
814            filter = dict_to_proto_struct(filter)
815        args_dict = self._parse_non_empty_args([("filter", filter)])
816        timeout = kwargs.pop("timeout", None)
817
818        request = DescribeIndexStatsRequest(**args_dict)
819        response = self._wrap_grpc_call(self.stub.DescribeIndexStats, request, timeout=timeout)
820        json_response = json_format.MessageToDict(response)
821        return parse_stats_response(json_response)
822
823    @staticmethod
824    def _parse_non_empty_args(args: List[Tuple[str, Any]]) -> Dict[str, Any]:
825        return {arg_name: val for arg_name, val in args if val is not None}
826
827    @staticmethod
828    def _parse_sparse_values_arg(
829        sparse_values: Optional[Union[GRPCSparseValues, Dict[str, Union[List[float], List[int]]]]]
830    ) -> Optional[GRPCSparseValues]:
831        if sparse_values is None:
832            return None
833
834        if isinstance(sparse_values, GRPCSparseValues):
835            return sparse_values
836
837        if not isinstance(sparse_values, dict) or "indices" not in sparse_values or "values" not in sparse_values:
838            raise ValueError(
839                "Invalid sparse values argument. Expected a dict of: {'indices': List[int], 'values': List[float]}."
840                f"Received: {sparse_values}"
841            )
842
843        return GRPCSparseValues(indices=sparse_values["indices"], values=sparse_values["values"])