"""Synchronous gRPC data plane client for a Pinecone index."""
from __future__ import annotations
import builtins
import logging
import os
import threading
import warnings
from collections.abc import Callable, Iterator, Mapping, Sequence
from concurrent.futures import ThreadPoolExecutor
from typing import TYPE_CHECKING, Any, Literal
from urllib.parse import quote
if TYPE_CHECKING:
import pandas as pd # type: ignore[import-untyped]
from pinecone._internal.adapters.imports_adapter import ImportsAdapter
from pinecone._internal.adapters.vectors_adapter import VectorsAdapter, extract_response_info
from pinecone._internal.adaptive import _AdaptiveLimiterRegistry
from pinecone._internal.batch import batch_execute
from pinecone._internal.batching import validate_batch_size
from pinecone._internal.config import PineconeConfig, RetryConfig
from pinecone._internal.constants import DATA_PLANE_API_VERSION
from pinecone._internal.data_plane_helpers import _build_search_records_body, _validate_host
from pinecone._internal.dataframe import _resolve_on_error, extract_records
from pinecone._internal.keyword_only import keyword_only_methods
from pinecone._internal.validation import (
DELETE_EMPTY_FILTER_MESSAGE,
FETCH_BY_METADATA_EMPTY_FILTER_MESSAGE,
QUERY_TOP_K_MAX,
UPDATE_EMPTY_FILTER_MESSAGE,
require_creatable_namespace_name,
require_delete_selectors,
require_in_range,
require_non_empty_filter,
require_query_selectors,
require_update_selectors,
require_valid_fetch_by_metadata_limit,
require_valid_id_prefix,
require_valid_list_limit,
require_valid_namespace_limit,
require_valid_namespace_name,
require_valid_namespace_prefix,
require_valid_namespace_schema,
require_valid_vector_id,
require_valid_vector_ids,
)
from pinecone._internal.vector_factory import VectorFactory, validate_vector_dict
from pinecone.errors.exceptions import (
PineconeTimeoutError,
PineconeValueError,
ValidationError,
)
from pinecone.grpc._protocol import GrpcChannelProtocol
from pinecone.grpc.future import PineconeFuture
from pinecone.models.batch import BatchResult
from pinecone.models.imports.list import ImportList
from pinecone.models.imports.model import ImportModel, StartImportResponse
from pinecone.models.namespaces.models import (
IndexedFields,
ListNamespacesResponse,
NamespaceDescription,
NamespaceFieldConfig,
NamespaceSchema,
)
from pinecone.models.vectors.query_aggregator import QueryNamespacesResults, QueryResultsAggregator
from pinecone.models.vectors.responses import (
DescribeIndexStatsResponse,
FetchByMetadataResponse,
FetchResponse,
ListItem,
ListResponse,
NamespaceSummary,
Pagination,
QueryResponse,
UpdateResponse,
UpsertRecordsResponse,
UpsertResponse,
)
from pinecone.models.vectors.search import (
RerankConfig,
SearchInputs,
SearchQuery,
SearchRecordsResponse,
)
from pinecone.models.vectors.sparse import SparseValues
from pinecone.models.vectors.usage import Usage
from pinecone.models.vectors.vector import ScoredVector, Vector
logger = logging.getLogger(__name__)
def _build_grpc_endpoint(host: str, secure: bool) -> str:
"""Build a gRPC endpoint URL from a host string.
Strips any existing scheme and applies the correct one for gRPC.
"""
bare = host
for prefix in ("https://", "http://"):
if bare.startswith(prefix):
bare = bare[len(prefix) :]
break
scheme = "https" if secure else "http"
return f"{scheme}://{bare}"
def _vector_to_grpc_dict(v: Vector) -> dict[str, Any]:
"""Serialize a Vector to a dict matching GrpcChannel's expected input format."""
d: dict[str, Any] = {"id": v.id, "values": v.values}
if v.sparse_values is not None:
d["sparse_values"] = {
"indices": v.sparse_values.indices,
"values": v.sparse_values.values,
}
if v.metadata is not None:
d["metadata"] = v.metadata
return d
def _dict_to_vector(vid: str, data: dict[str, Any]) -> Vector:
"""Convert a GrpcChannel vector dict to a Vector model."""
sparse = None
sv = data.get("sparse_values")
if sv is not None:
sparse = SparseValues(sv["indices"], sv["values"])
return Vector(
id=vid,
values=data.get("values", []),
sparse_values=sparse,
metadata=data.get("metadata"),
)
def _dict_to_scored_vector(data: dict[str, Any]) -> ScoredVector:
"""Convert a GrpcChannel scored vector dict to a ScoredVector model."""
sparse = None
sv = data.get("sparse_values")
if sv is not None:
sparse = SparseValues(sv["indices"], sv["values"])
return ScoredVector(
id=data["id"],
score=data.get("score", 0.0),
values=data.get("values", []),
sparse_values=sparse,
metadata=data.get("metadata"),
)
def _dict_to_usage(data: dict[str, Any] | None) -> Usage | None:
"""Convert a usage dict to a Usage model, or None."""
if data is None:
return None
return Usage(read_units=data.get("read_units", 0))
def _dict_to_namespace_description(data: dict[str, Any]) -> NamespaceDescription:
"""Convert a GrpcChannel namespace dict to a NamespaceDescription model.
Shared by create_namespace, describe_namespace, and list_namespaces_paginated
to convert the dict payload returned by the Rust-backed GrpcChannel into a
typed NamespaceDescription, including optional schema and indexed_fields.
``indexed_fields`` arrives as a bare list of names here, where the REST JSON
nests the same names under a ``fields`` key — see
``namespace_description_to_py_dict`` in rust/src/transport.rs. Both shapes
have to produce the same model, so the two readers cannot be collapsed.
"""
schema: NamespaceSchema | None = None
raw_schema = data.get("schema")
if raw_schema is not None:
schema = NamespaceSchema(
fields={
k: NamespaceFieldConfig(filterable=v["filterable"])
for k, v in raw_schema.get("fields", {}).items()
}
)
indexed_fields: IndexedFields | None = None
raw_indexed = data.get("indexed_fields")
if raw_indexed is not None:
indexed_fields = IndexedFields(fields=list(raw_indexed))
return NamespaceDescription(
name=data.get("name", ""),
record_count=data.get("record_count", 0),
schema=schema,
indexed_fields=indexed_fields,
size_bytes=data.get("size_bytes", 0),
)
# gRPC's retry defaults, which differ from REST's (3 / 0.25s / 60s). The counts and
# floor are what the Rust layer has always used; only the cap changed, from 1.6s — a
# value small enough to swallow a `grpc-retry-pushback-ms: 30000` hint from the server.
_GRPC_DEFAULT_MAX_RETRIES = 5
_GRPC_DEFAULT_BACKOFF_FACTOR = 0.1
_GRPC_DEFAULT_MAX_WAIT = 60.0
_warned_about_grpc_partial_failure = False
def _warn_grpc_partial_failure_once(response: UpsertResponse) -> None:
"""Announce the 9.2.0 change the first time a caller is affected by it.
Only on gRPC: REST has aggregated since v9.0.0, so warning there would be
noise about behavior that did not change.
"""
global _warned_about_grpc_partial_failure
if _warned_about_grpc_partial_failure:
return
_warned_about_grpc_partial_failure = True
warnings.warn(
f"{response.failed_item_count} of {response.total_item_count} vectors failed to "
"upsert. As of 9.2.0 upsert_from_dataframe aggregates partial failures instead "
"of raising: inspect response.errors and retry response.failed_items. Pass "
'on_error="raise" to restore the previous behavior, or on_error="collect" to '
"silence this warning.",
stacklevel=3,
)
def _upsert_response_from(batch_result: BatchResult) -> UpsertResponse:
"""Project a BatchResult onto the response shape callers already handle."""
return UpsertResponse(
upserted_count=batch_result.successful_item_count,
total_item_count=batch_result.total_item_count,
failed_item_count=batch_result.failed_item_count,
total_batch_count=batch_result.total_batch_count,
successful_batch_count=batch_result.successful_batch_count,
failed_batch_count=batch_result.failed_batch_count,
errors=batch_result.errors,
)
def _limiter_host(host: str) -> str:
"""The key the Rust throttle callback reports under.
``self._host`` carries a scheme, because ``normalize_host`` adds one; the
callback receives what ``parse_host_from_endpoint`` produced, which is the
bare hostname. Registering a limiter under one and reporting throttles
against the other would leave the limiter permanently at its ceiling.
"""
bare = host
for prefix in ("https://", "http://"):
if bare.startswith(prefix):
bare = bare[len(prefix) :]
break
for separator in (":", "/"):
bare = bare.split(separator, 1)[0]
return bare
def _default_max_concurrency() -> int:
"""What an unbounded submission into a default ThreadPoolExecutor already gave.
Bounding submission without keeping this number would be a throughput
regression rather than a fix, so it is the default and callers override it.
"""
return min(32, (os.cpu_count() or 1) + 4)
[docs]
@keyword_only_methods
class GrpcIndex:
"""Synchronous gRPC data plane client targeting a specific Pinecone index.
Provides the same interface as :class:`~pinecone.index.Index` but routes
data-plane operations through a gRPC transport (via the Rust-backed
:class:`~pinecone._grpc.GrpcChannel`) instead of HTTP/REST.
Args:
host (str): The index-specific data plane host URL.
api_key (str | None): Pinecone API key. Falls back to ``PINECONE_API_KEY`` env var.
api_version (str): API version string. Defaults to the current data plane version.
source_tag (str | None): Tag appended to the User-Agent string for request attribution.
secure (bool): Whether to use TLS encryption. Defaults to ``True``.
timeout (float): Request timeout in seconds. Defaults to ``20.0``.
connect_timeout (float): Connection timeout in seconds. Defaults to ``1.0``.
retry_config (RetryConfig | None): Retry policy for transient gRPC errors. Accepts
the same :class:`~pinecone._internal.config.RetryConfig` REST uses. ``None``
(default) uses the gRPC defaults: ``max_retries=5``, ``backoff_factor=0.1``,
``max_wait=60.0``. ``retryable_status_codes`` is **ignored on this transport** —
it carries HTTP statuses, while gRPC retries a fixed set of ``tonic::Code``
values (UNAVAILABLE, RESOURCE_EXHAUSTED, ABORTED).
proxy_url (str | None): HTTP proxy URL. gRPC traffic is tunnelled through it with
HTTP CONNECT.
limiter_registry (_AdaptiveLimiterRegistry | None): SDK-internal. Registry the
bulk paths consult to back off under throttling. Wired by
:meth:`Pinecone.index`; not intended for user configuration.
Raises:
:exc:`PineconeValueError`: If no API key can be resolved or the host is invalid.
Note:
**Four timeout layers apply to every gRPC call**, and only the first three bound a
single request:
1. **Connect** — ``connect_timeout``, default ``1.0s``.
2. **Per attempt** — ``timeout`` (or a per-call ``timeout=``), default ``20.0s``.
This is a deadline on *one attempt*, not on the call.
3. **Retry budget** — ``retry_config.max_retries`` attempts after the first, with
backoff between them.
4. **Whole job** — for bulk methods only, ``total_timeout``.
Layers 2 and 3 multiply. ``timeout=120`` is not a 120s bound: with the default
``max_retries=5`` it is up to 6 attempts × 120s **plus** backoff, so a worst case
near 17 minutes. Lower ``max_retries`` to shrink that, or bound the whole
operation with ``total_timeout``.
Examples:
.. code-block:: python
from pinecone.grpc import GrpcIndex
idx = GrpcIndex(host="movie-recs-abc123.svc.pinecone.io", api_key="...")
"""
[docs]
def __init__(
self,
*,
host: str,
api_key: str | None = None,
api_version: str = DATA_PLANE_API_VERSION,
source_tag: str | None = None,
secure: bool = True,
timeout: float = 20.0,
connect_timeout: float = 1.0,
retry_config: RetryConfig | None = None,
proxy_url: str | None = None,
on_throttle: Callable[[str], None] | None = None,
limiter_registry: _AdaptiveLimiterRegistry | None = None,
) -> None:
# Resolve API key: explicit arg > env var
resolved_key = api_key or os.environ.get("PINECONE_API_KEY", "")
if not resolved_key:
raise ValidationError(
"No API key provided. Pass api_key='...' or set the "
"PINECONE_API_KEY environment variable."
)
# Validate and normalize host
self._host = _validate_host(host)
self._limiter_host = _limiter_host(self._host)
self._limiter_registry = limiter_registry
self._source_tag = source_tag
# Build gRPC endpoint and create the Rust-backed channel
endpoint = _build_grpc_endpoint(self._host, secure)
from pinecone import __version__
from pinecone._grpc import GrpcChannel # type: ignore[import-not-found]
# `retryable_status_codes` is deliberately not forwarded: it carries HTTP
# statuses, and this transport retries a fixed set of tonic::Code values.
# Forcing HTTP statuses through a gRPC channel would be meaningless.
self._retry_config = retry_config or RetryConfig(
max_retries=_GRPC_DEFAULT_MAX_RETRIES,
backoff_factor=_GRPC_DEFAULT_BACKOFF_FACTOR,
max_wait=_GRPC_DEFAULT_MAX_WAIT,
)
# RetryConfig.on_throttle is how REST carries the limiter hook; honor it when
# the explicit argument is absent so threading a client-built config does not
# silently drop the callback.
resolved_on_throttle = on_throttle or self._retry_config.on_throttle
self._channel: GrpcChannelProtocol = GrpcChannel(
endpoint,
resolved_key,
api_version,
__version__,
secure,
timeout,
connect_timeout,
max_retries=self._retry_config.max_retries,
backoff_factor_s=self._retry_config.backoff_factor,
max_wait_s=self._retry_config.max_wait,
source_tag=source_tag,
proxy_url=proxy_url,
on_throttle=resolved_on_throttle,
)
self._executor = ThreadPoolExecutor()
self._batch_executors: dict[int, ThreadPoolExecutor] = {}
self._batch_executor_lock = threading.Lock()
# REST HTTP client for records operations (integrated inference).
# upsert_records and search use REST endpoints with no gRPC equivalent.
from pinecone._internal.http_client import HTTPClient
rest_config = PineconeConfig(
api_key=resolved_key,
host=self._host,
timeout=timeout,
source_tag=source_tag or "",
ssl_verify=secure,
)
self._http = HTTPClient(rest_config, DATA_PLANE_API_VERSION)
self._adapter = VectorsAdapter()
self._imports_adapter = ImportsAdapter()
logger.info("GrpcIndex client created for host %s", self._host)
@property
def host(self) -> str:
"""The data plane host URL for this index."""
return self._host
def _get_batch_executor(self, max_concurrency: int) -> ThreadPoolExecutor:
"""Return the pool of this size, creating it once.
Pools are kept per size rather than resized in place. upsert() and
upsert_from_dataframe() have different concurrency defaults and can run
concurrently on one index handle; shutting a pool down because the other
caller asked for a different size would raise "cannot schedule new
futures after shutdown" in whichever one was still submitting.
"""
with self._batch_executor_lock:
executor = self._batch_executors.get(max_concurrency)
if executor is None:
executor = ThreadPoolExecutor(
max_concurrency,
thread_name_prefix="pinecone-grpc-batch-upsert",
)
self._batch_executors[max_concurrency] = executor
return executor
[docs]
def upsert(
self,
*,
vectors: Sequence[
Vector
| tuple[str, Sequence[float]]
| tuple[str, Sequence[float], Mapping[str, Any]]
| Mapping[str, Any]
],
namespace: str = "",
batch_size: int | None = None,
max_concurrency: int = 4,
show_progress: bool = True,
timeout: float | None = None,
) -> UpsertResponse:
"""Upsert a batch of vectors into a namespace.
If a vector with the same ID already exists in the namespace, it is
overwritten.
One request is capped both on the number of vectors it carries and on
its encoded size, and with wide vectors or heavy metadata the size cap
is usually the one reached first. Pass ``batch_size`` to split a long
sequence into requests that stay under both.
Args:
vectors: Sequence of vectors to upsert. Each element can be a
``Vector`` instance, a tuple of ``(id, values)`` or
``(id, values, metadata)``, or a dict with ``id``, ``values``,
and optional ``sparse_values`` / ``metadata`` keys.
namespace (str): Target namespace. Defaults to the default
(empty-string) namespace.
batch_size (int | None): If set, splits ``vectors`` into batches of
this size and submits them in **parallel**. ``None`` (default)
sends all vectors in a single request. Must be a positive
integer when set.
max_concurrency (int): Number of parallel threads used when
``batch_size`` is set. Default ``4``, range ``[1, 64]``. Ignored
when ``batch_size`` is ``None``.
show_progress (bool): If ``True`` and ``tqdm`` is installed, display a
progress bar while submitting batches. Ignored when ``batch_size``
is ``None``. Defaults to ``True``.
timeout (float | None): Per-call timeout in seconds. Applied per batch
when batching. None uses the client-level default.
Returns:
:class:`UpsertResponse` with the count of vectors upserted.
Raises:
:exc:`PineconeTypeError`: If a vector element is not a recognized format.
:exc:`PineconeValueError`: If a vector element is malformed, if
``batch_size`` is not a positive integer, or if
``max_concurrency`` is outside ``[1, 64]``.
:exc:`ApiError`: If one request exceeds the server's cap on vectors
per request or on encoded request size. Lower ``batch_size``
and retry.
:exc:`PineconeTimeoutError`: If the call does not complete before
*timeout* elapses.
Notes:
When ``batch_size`` is set, up to ``max_concurrency`` batches run
at once (default 4, range 1-64), each retried independently on
transient errors. **Partial failures do not raise** — the
returned :class:`UpsertResponse` carries ``upserted_count``,
``failed_item_count``, ``errors``, and ``failed_items`` for
inspection or retry. Pass ``response.failed_items`` back to
``upsert(...)`` to retry only the failures.
Examples:
.. code-block:: python
from pinecone.grpc import GrpcIndex
from pinecone.models.vectors.vector import Vector
idx = GrpcIndex(host="article-search-abc123.svc.pinecone.io", api_key="...")
response = idx.upsert(
vectors=[
Vector(
id="article-101",
values=[0.012, -0.087, 0.153, ...], # 1536-dim
),
("article-102", [0.045, 0.021, -0.064, ...]),
{"id": "article-103", "values": [0.091, -0.032, 0.178, ...]},
],
namespace="articles-en",
)
print(response.upserted_count)
"""
if batch_size is None:
built = [VectorFactory.build(v) for v in vectors]
grpc_vectors = [_vector_to_grpc_dict(v) for v in built]
logger.info("Upserting %d vectors via gRPC into namespace %r", len(built), namespace)
result = self._channel.upsert(grpc_vectors, namespace or None, timeout_s=timeout)
return UpsertResponse(upserted_count=result.get("upserted_count", 0))
validate_batch_size(batch_size)
require_in_range("max_concurrency", max_concurrency, 1, 64)
built = [VectorFactory.build(v) for v in vectors]
items: builtins.list[dict[str, Any]] = [_vector_to_grpc_dict(v) for v in built]
def _operation(chunk: builtins.list[dict[str, Any]]) -> dict[str, Any]:
return self._channel.upsert(chunk, namespace or None, timeout_s=timeout)
batch_result = batch_execute(
items=items,
operation=_operation,
batch_size=batch_size,
max_concurrency=max_concurrency,
show_progress=show_progress,
desc="Upserting",
executor=self._get_batch_executor(max_concurrency),
limiter_registry=self._limiter_registry,
host=self._limiter_host,
)
return UpsertResponse(
upserted_count=batch_result.successful_item_count,
total_item_count=batch_result.total_item_count,
failed_item_count=batch_result.failed_item_count,
total_batch_count=batch_result.total_batch_count,
successful_batch_count=batch_result.successful_batch_count,
failed_batch_count=batch_result.failed_batch_count,
errors=batch_result.errors,
)
[docs]
def query(
self,
*,
top_k: int,
vector: Sequence[float] | None = None,
id: str | None = None,
namespace: str = "",
filter: Mapping[str, Any] | None = None,
include_values: bool = False,
include_metadata: bool = False,
sparse_vector: SparseValues | Mapping[str, Any] | None = None,
scan_factor: float | None = None,
max_candidates: int | None = None,
timeout: float | None = None,
) -> QueryResponse:
"""Query a namespace for the nearest neighbors of a vector.
.. note::
Vector operations remain available for indexes created before 2026-07,
where you supply your own vectors. An index created at 2026-07 carries
a document schema instead, and its reads and writes go through the
document operations.
Args:
top_k (int): Number of results to return, 1-10000.
vector (list[float] | None): Dense query vector values.
id (str | None): ID of a stored vector to use as the query.
namespace (str): Namespace to query. Defaults to the default namespace.
filter (dict[str, Any] | None): Metadata filter expression.
include_values (bool): Whether to include vector values in results.
include_metadata (bool): Whether to include metadata in results.
sparse_vector (SparseValues | dict[str, Any] | None): Sparse query vector
with indices and values.
scan_factor (float | None): Recall/latency trade for dedicated read
node (DRN) indexes — a multiplier on how much of the index is
scanned. Above 1 scans more and favours recall; below 1 scans
less and favours latency. Omit to let the server choose.
max_candidates (int | None): Recall/latency trade for dedicated read
node (DRN) indexes — caps how many candidates are reranked before
``top_k`` is taken. Must be at least ``top_k``: a smaller value is
rejected rather than clamped, since it could not fill the page.
timeout (float | None): Per-call timeout in seconds. None uses the client-level default.
Returns:
:class:`QueryResponse` with matches, namespace, and usage info.
Raises:
:exc:`PineconeValueError`: If top_k is not between 1 and 10000, ``id``
is combined with ``vector`` or ``sparse_vector``, none of
``vector``, ``id``, or ``sparse_vector`` is provided, or ``id``
is not a legal vector ID.
:exc:`ApiError`: If ``scan_factor`` or ``max_candidates`` is out of
range, or the index is not a dense DRN index — both knobs are
rejected on on-demand indexes and on sparse indexes.
:exc:`PineconeTimeoutError`: If the call does not complete before
*timeout* elapses.
Examples:
.. code-block:: python
response = idx.query(
top_k=10,
vector=[0.012, -0.087, 0.153, ...], # 1536-dim embedding
)
for match in response.matches:
print(match.id, match.score)
"""
require_in_range("top_k", top_k, 1, QUERY_TOP_K_MAX)
require_query_selectors(vector=vector, id=id, sparse_vector=sparse_vector)
if id is not None:
require_valid_vector_id("id", id)
# Convert SparseValues model to dict for GrpcChannel
sv_dict: Mapping[str, Any] | None = None
if sparse_vector is not None:
if isinstance(sparse_vector, SparseValues):
sv_dict = {
"indices": sparse_vector.indices,
"values": sparse_vector.values,
}
else:
sv_dict = sparse_vector
logger.info("Querying index via gRPC with top_k=%d", top_k)
result = self._channel.query(
top_k,
vector=vector,
id=id,
namespace=namespace or None,
filter=filter,
include_values=include_values,
include_metadata=include_metadata,
sparse_vector=sv_dict,
scan_factor=scan_factor,
max_candidates=max_candidates,
timeout_s=timeout,
)
matches = [_dict_to_scored_vector(m) for m in result.get("matches", [])]
usage = _dict_to_usage(result.get("usage"))
return QueryResponse(
matches=matches,
namespace=result.get("namespace", ""),
usage=usage,
)
[docs]
def query_namespaces(
self,
*,
vector: Sequence[float] | None = None,
namespaces: Sequence[str],
metric: str,
top_k: int | None = None,
filter: Mapping[str, Any] | None = None,
include_values: bool = False,
include_metadata: bool = False,
sparse_vector: SparseValues | Mapping[str, Any] | None = None,
scan_factor: float | None = None,
max_candidates: int | None = None,
timeout: float | None = None,
) -> QueryNamespacesResults:
"""Query multiple namespaces in parallel and return merged top results.
Fans out individual ``query()`` calls across all given namespaces
using a thread pool, then merges results via a heap-based aggregator
that returns the overall top-k matches ranked by the specified metric.
Args:
vector: Dense query vector values. Required for dense and hybrid
indexes; omit for sparse-only indexes (use *sparse_vector* instead).
namespaces: Namespaces to query (must be non-empty). Duplicates
are removed while preserving order.
metric: Distance metric — ``"cosine"``, ``"euclidean"``, or
``"dotproduct"``.
top_k: Maximum number of results to return. Defaults to 10.
filter: Metadata filter expression applied to every namespace.
include_values: Whether to include vector values in results.
include_metadata: Whether to include metadata in results.
sparse_vector: Sparse query vector with indices and values.
Required for sparse-only indexes when *vector* is omitted.
scan_factor: Recall/latency trade for dedicated read node (DRN)
indexes — a multiplier on how much of the index is scanned.
Above 1 scans more and favours recall; below 1 scans less and
favours latency. Applied to every namespace queried.
max_candidates: Recall/latency trade for dedicated read node (DRN)
indexes — caps how many candidates are reranked before ``top_k``
is taken, per namespace. Must be at least ``top_k``.
Returns:
:class:`QueryNamespacesResults` with the merged top-k matches, total
usage, and per-namespace usage.
Raises:
:exc:`PineconeValueError`: If *namespaces* is empty, if both
*vector* and *sparse_vector* are absent/empty, or if *metric*
is not a recognized value.
:exc:`ApiError`: If any individual namespace query fails.
:exc:`PineconeConnectionError`: If a network-level connection
fails (DNS, refused, transport error).
:exc:`PineconeTimeoutError`: If the request does not complete
before the configured timeout elapses.
Examples:
.. code-block:: python
# Dense query
results = idx.query_namespaces(
vector=[0.012, -0.087, 0.153], # truncated; use your actual dimension
namespaces=["articles-en", "articles-fr", "articles-de"],
metric="cosine",
top_k=10,
)
# Sparse-only query (sparse index)
results = idx.query_namespaces(
sparse_vector={"indices": [0, 1, 2], "values": [0.1, 0.2, 0.3]},
namespaces=["docs-en", "docs-fr"],
metric="dotproduct",
top_k=10,
)
for match in results.matches:
print(match.id, match.score)
"""
if not namespaces:
raise ValidationError("namespaces must be a non-empty list")
if not vector and not sparse_vector:
raise ValidationError("at least one of 'vector' or 'sparse_vector' must be provided")
valid_metrics = {"cosine", "euclidean", "dotproduct"}
if metric not in valid_metrics:
raise ValidationError(
f"Invalid metric {metric!r}. Must be one of: {', '.join(sorted(valid_metrics))}"
)
namespaces = list(dict.fromkeys(namespaces))
effective_top_k = top_k if top_k is not None else 10
aggregator = QueryResultsAggregator(metric=metric, top_k=effective_top_k)
query_kwargs: dict[str, Any] = {
"top_k": effective_top_k,
"filter": filter,
"include_values": include_values,
"include_metadata": include_metadata,
"sparse_vector": sparse_vector,
"scan_factor": scan_factor,
"max_candidates": max_candidates,
"timeout": timeout,
}
if vector is not None:
query_kwargs["vector"] = vector
with ThreadPoolExecutor(max_workers=min(len(namespaces), 32)) as pool:
futures = [pool.submit(self.query, namespace=ns, **query_kwargs) for ns in namespaces]
for ns, future in zip(namespaces, futures, strict=True):
aggregator.add_results(ns, future.result())
return aggregator.get_results()
[docs]
def fetch(
self,
*,
ids: Sequence[str],
namespace: str = "",
timeout: float | None = None,
) -> FetchResponse:
"""Fetch vectors by their IDs from a namespace.
Args:
ids (list[str]): List of vector IDs to fetch (must be non-empty).
namespace (str): Namespace to fetch from. Defaults to the default namespace.
timeout (float | None): Per-call timeout in seconds. None uses the client-level default.
Returns:
:class:`FetchResponse` with a map of vector IDs to Vector objects, namespace,
and usage info.
Raises:
:exc:`PineconeValueError`: If ids is empty or any ID is not 1-512
ASCII characters without a NUL.
:exc:`PineconeTimeoutError`: If the call does not complete before
*timeout* elapses.
Examples:
.. code-block:: python
response = idx.fetch(ids=["article-101", "article-102"])
for vid, vec in response.vectors.items():
print(vid, vec.values)
"""
require_valid_vector_ids("ids", ids)
logger.info("Fetching %d vectors via gRPC", len(ids))
result = self._channel.fetch(ids, namespace=namespace or None, timeout_s=timeout)
vectors: dict[str, Vector] = {}
for vid, vdata in result.get("vectors", {}).items():
vectors[vid] = _dict_to_vector(vid, vdata)
usage = _dict_to_usage(result.get("usage"))
return FetchResponse(
vectors=vectors,
namespace=result.get("namespace", ""),
usage=usage,
)
[docs]
def delete(
self,
*,
ids: Sequence[str] | None = None,
delete_all: bool = False,
filter: Mapping[str, Any] | None = None,
namespace: str = "",
timeout: float | None = None,
) -> None:
"""Delete vectors from a namespace by ID, filter, or delete-all flag.
Exactly one of ``ids``, ``delete_all``, or ``filter`` must be specified.
A by-filter delete selects on metadata alone, so a text-match operator
(``$match_phrase``, ``$match_all``, ``$match_any``) in the filter is
rejected rather than ignored — evaluated there it would match everything
and widen the delete to every record the rest of the filter admits. Text
matching belongs in :meth:`search`.
A by-filter delete also reads before it writes, so a dedicated index
scaled to zero replicas refuses it; add replicas first. Deleting by ID or
with ``delete_all`` is unaffected.
Args:
ids (list[str] | None): List of vector IDs to delete.
delete_all (bool): If True, delete all vectors in the namespace.
filter (dict[str, Any] | None): Metadata filter expression selecting vectors to delete.
namespace (str): Namespace to delete from. Defaults to the default namespace.
timeout (float | None): Per-call timeout in seconds. None uses the client-level default.
Returns:
None
Raises:
:exc:`PineconeValueError`: If zero or more than one deletion mode is
specified, any ID is not a legal vector ID, or ``filter`` is empty.
:exc:`ApiError`: If a by-filter delete uses a text-match operator, or
the index is a dedicated index scaled to zero replicas.
:exc:`PineconeTimeoutError`: If the call does not complete before
*timeout* elapses.
Examples:
.. code-block:: python
# Delete by IDs
idx.delete(ids=["article-101", "article-102"])
# Delete all vectors in a namespace
idx.delete(delete_all=True, namespace="articles-deprecated")
# Delete by metadata filter
idx.delete(filter={"category": {"$eq": "obsolete"}})
"""
require_delete_selectors(ids=ids, delete_all=delete_all, filter=filter)
if ids is not None:
require_valid_vector_ids("ids", ids)
if filter is not None:
require_non_empty_filter("filter", filter, server_message=DELETE_EMPTY_FILTER_MESSAGE)
logger.info("Deleting vectors via gRPC from namespace %r", namespace)
self._channel.delete(
ids=ids,
delete_all=delete_all,
namespace=namespace or None,
filter=filter,
timeout_s=timeout,
)
[docs]
def update(
self,
*,
id: str | None = None,
values: Sequence[float] | None = None,
sparse_values: SparseValues | Mapping[str, Any] | None = None,
set_metadata: Mapping[str, Any] | None = None,
namespace: str = "",
filter: Mapping[str, Any] | None = None,
dry_run: bool = False,
timeout: float | None = None,
) -> UpdateResponse:
"""Update vectors by ID or metadata filter.
A by-filter update selects on metadata alone, so a text-match operator
(``$match_phrase``, ``$match_all``, ``$match_any``) in the filter is
rejected rather than ignored — evaluated there it would match everything
and widen the patch to every record the rest of the filter admits. Text
matching belongs in :meth:`search`.
A by-filter update also reads before it writes, so a dedicated index
scaled to zero replicas refuses it; add replicas first. Updating by ID is
unaffected.
Args:
id (str | None): ID of the vector to update.
values (list[float] | None): New dense vector values.
sparse_values (SparseValues | dict[str, Any] | None): New sparse vector.
set_metadata (dict[str, Any] | None): Metadata fields to set or overwrite.
namespace (str): Namespace to target. Defaults to the default namespace.
filter (dict[str, Any] | None): Metadata filter expression selecting vectors to update.
dry_run (bool): If True, return the count of records that would be
affected without applying changes.
timeout (float | None): Per-call timeout in seconds. None uses the client-level default.
Returns:
:class:`UpdateResponse` with matched_records count (when available).
Raises:
:exc:`PineconeValueError`: If both or neither of id and filter are
provided, if ``filter`` is combined with ``values`` or
``sparse_values``, if ``filter`` is empty, or if ``id`` is not
a legal vector ID.
:exc:`ApiError`: If a by-filter update uses a text-match operator, or
the index is a dedicated index scaled to zero replicas.
:exc:`PineconeTimeoutError`: If the call does not complete before
*timeout* elapses.
Examples:
.. code-block:: python
# Update by ID
idx.update(id="article-101", values=[0.012, -0.087, 0.153, ...])
# Bulk-update metadata by filter
idx.update(
filter={"genre": {"$eq": "drama"}},
set_metadata={"year": 2020},
)
"""
require_update_selectors(id=id, filter=filter, values=values, sparse_values=sparse_values)
if id is not None:
require_valid_vector_id("id", id)
if filter is not None:
require_non_empty_filter("filter", filter, server_message=UPDATE_EMPTY_FILTER_MESSAGE)
# Convert SparseValues model to dict for GrpcChannel
sv_dict: Mapping[str, Any] | None = None
if sparse_values is not None:
if isinstance(sparse_values, SparseValues):
sv_dict = {
"indices": sparse_values.indices,
"values": sparse_values.values,
}
else:
sv_dict = sparse_values
logger.info("Updating vectors via gRPC in namespace %r", namespace)
# The Rust channel's update() requires `id` as a positional string arg.
# For filter-based updates id is None, so pass "" which the API ignores
# when a filter is provided.
result = self._channel.update(
id if id is not None else "",
values=values,
sparse_values=sv_dict,
set_metadata=set_metadata,
namespace=namespace or None,
filter=filter,
dry_run=dry_run or None,
timeout_s=timeout,
)
return UpdateResponse(matched_records=result.get("matched_records"))
[docs]
def list_paginated(
self,
*,
prefix: str | None = None,
limit: int | None = None,
pagination_token: str | None = None,
namespace: str = "",
timeout: float | None = None,
) -> ListResponse:
"""Fetch a single page of vector IDs from a namespace.
Args:
prefix (str | None): Return only IDs starting with this prefix.
limit (int | None): Maximum number of IDs to return in this page, 1-100.
pagination_token (str | None): Token from a previous response to fetch the next page.
namespace (str): Namespace to list from. Defaults to the default namespace.
timeout (float | None): Per-call timeout in seconds. None uses the client-level default.
Returns:
:class:`ListResponse` with vector IDs, pagination info, namespace, and usage.
Raises:
:exc:`PineconeValueError`: If ``prefix`` is not legal or ``limit``
falls outside 1-100.
:exc:`PineconeTimeoutError`: If the call does not complete before
*timeout* elapses.
Examples:
.. code-block:: python
response = idx.list_paginated(prefix="doc1#", limit=50)
for item in response.vectors:
print(item.id)
"""
if prefix is not None:
require_valid_id_prefix("prefix", prefix)
if limit is not None:
require_valid_list_limit("limit", limit)
logger.info("Listing vectors via gRPC in namespace %r", namespace)
result = self._channel.list(
prefix=prefix,
limit=limit,
pagination_token=pagination_token,
namespace=namespace or None,
timeout_s=timeout,
)
vectors = [ListItem(id=v.get("id")) for v in result.get("vectors", [])]
pagination_data = result.get("pagination")
pagination = None
if pagination_data is not None:
pagination = Pagination(next=pagination_data.get("next"))
usage = _dict_to_usage(result.get("usage"))
return ListResponse(
vectors=vectors,
pagination=pagination,
namespace=result.get("namespace", ""),
usage=usage,
)
[docs]
def list(
self,
*,
prefix: str | None = None,
limit: int | None = None,
namespace: str = "",
timeout: float | None = None,
) -> Iterator[ListResponse]:
"""List vector IDs in a namespace, automatically following pagination.
Yields one ``ListResponse`` per page.
Args:
prefix (str | None): Return only IDs starting with this prefix.
limit (int | None): Maximum number of IDs to return per page.
namespace (str): Namespace to list from. Defaults to the default namespace.
timeout (float | None): Per-call timeout in seconds applied to each page
request. None uses the client-level default.
Yields:
:class:`ListResponse` for each page of results.
Raises:
:exc:`PineconeValueError`: If ``prefix`` is not legal or ``limit``
falls outside 1-100.
:exc:`PineconeTimeoutError`: If a page request does not complete
before *timeout* elapses.
Examples:
.. code-block:: python
for page in idx.list(prefix="doc1#"):
for item in page.vectors:
print(item.id)
"""
pagination_token: str | None = None
while True:
page = self.list_paginated(
prefix=prefix,
limit=limit,
pagination_token=pagination_token,
namespace=namespace,
timeout=timeout,
)
if page.vectors:
yield page
if page.pagination is not None and page.pagination.next is not None:
pagination_token = page.pagination.next
else:
break
[docs]
def describe_index_stats(
self,
*,
filter: Mapping[str, Any] | None = None,
timeout: float | None = None,
) -> DescribeIndexStatsResponse:
"""Return statistics for this index.
Args:
filter (dict[str, Any] | None): Metadata filter expression. Accepted
for API compatibility, but a non-empty filter is rejected for
every index type, so the call fails instead of returning
filtered counts. Leave it unset: the statistics returned always
describe the whole index.
timeout (float | None): Per-call timeout in seconds. None uses the
client-level default.
Returns:
:class:`DescribeIndexStatsResponse` with namespace summaries, dimension,
total vector count, and fullness metrics.
Raises:
:exc:`ApiError`: If a non-empty ``filter`` is provided, since it is
rejected for every index type.
:exc:`PineconeTimeoutError`: If the call does not complete before
*timeout* elapses.
Examples:
.. code-block:: python
stats = idx.describe_index_stats()
print(stats.total_vector_count, stats.dimension)
"""
logger.info("Describing index stats via gRPC")
result = self._channel.describe_index_stats(filter=filter, timeout_s=timeout)
namespaces: dict[str, NamespaceSummary] = {}
for ns_name, ns_data in result.get("namespaces", {}).items():
namespaces[ns_name] = NamespaceSummary(
vector_count=ns_data.get("vector_count", 0),
)
return DescribeIndexStatsResponse(
namespaces=namespaces,
dimension=result.get("dimension"),
index_fullness=result.get("index_fullness", 0.0),
total_vector_count=result.get("total_vector_count", 0),
metric=result.get("metric"),
vector_type=result.get("vector_type"),
memory_fullness=result.get("memory_fullness"),
storage_fullness=result.get("storage_fullness"),
)
[docs]
def upsert_from_dataframe(
self,
df: pd.DataFrame,
namespace: str = "",
batch_size: int = 500,
show_progress: bool = True,
timeout: float | None = None,
*,
max_concurrency: int | None = None,
total_timeout: float | None = None,
on_error: Literal["raise", "collect"] | None = None,
) -> UpsertResponse:
"""Upsert vectors from a pandas DataFrame.
Splits the DataFrame into batches of ``batch_size`` rows, submits
batches in parallel, and aggregates the results into a single
response.
Args:
df: A ``pandas.DataFrame`` with at least ``id`` and ``values``
columns. ``sparse_values`` and ``metadata`` columns are
included when present and non-None.
namespace: Target namespace. Defaults to the default namespace.
batch_size: Number of rows per upsert batch. Defaults to 500.
show_progress: If ``True`` and ``tqdm`` is installed, display a
progress bar. The bar advances as batches *complete*. If ``tqdm``
is not installed, silently falls back to no progress bar.
max_concurrency: Number of batches in flight at once, range
``[1, 64]``. ``None`` (default) uses ``min(32, cpu_count + 4)``
— pass a value to make throughput reproducible across hosts.
on_error: What to do when some batches fail. ``"collect"`` returns an
:class:`UpsertResponse` carrying ``failed_item_count``, ``errors``
and ``failed_items``, so the caller can retry only what failed —
the same contract the REST client has had since v9.0.0.
``"raise"`` re-raises the lowest-indexed batch failure, after all
batches have settled, with the partial result attached to the
exception's ``response`` attribute. ``None`` (default) behaves as
``"collect"`` and additionally warns once per process when a
partial failure occurs, since this method used to raise; pass
``"collect"`` explicitly to silence that.
total_timeout: Deadline in seconds for the **whole ingest**, as opposed
to *timeout*, which bounds a single attempt of a single batch. On
expiry no further batches are submitted; batches already in flight
are allowed to settle rather than being abandoned, since dropping
them client-side would not stop the server from applying them.
:exc:`PineconeTimeoutError` is then raised carrying the partial
:class:`UpsertResponse` on its ``response`` attribute, whose
``failed_items`` are the rows that were never sent. ``None``
(default) means the ingest is bounded only by the per-batch
deadlines.
timeout: Server-side deadline in seconds applied to *each batch's*
upsert request — not to the DataFrame as a whole. ``None``
(default) uses the client's configured request timeout: the
``timeout`` passed to :class:`GrpcIndex` (``20.0s`` unless you
override it). Each attempt of a batch is bounded by this
deadline (transient errors may be retried, so a batch's total
wall-clock can exceed it). Result collection then waits for the
server, so a large ingest is bounded only by these per-batch
deadlines rather than failing prematurely. Raise *timeout* to
give slow batches more time on the server.
**This is not a bound on the batch.** With the default
``max_retries=5`` a batch is up to 6 attempts × *timeout*, plus
backoff between them — ``timeout=120`` admits a worst case near 17
minutes for a single batch. See the four timeout layers on
:class:`GrpcIndex`. To shrink the multiplier, pass a
``retry_config`` with a lower ``max_retries`` when constructing the
index.
Returns:
:class:`UpsertResponse` with the total count of vectors upserted across
all batches.
Raises:
:exc:`RuntimeError`: If ``pandas`` is not installed. It is not an SDK
dependency; install it yourself with ``pip install pandas``.
:exc:`PineconeValueError`: If *df* is not a ``pandas.DataFrame`` or
*batch_size* is not a positive integer.
:exc:`PineconeTimeoutError`: If a batch exceeds *timeout* on the server,
or if *total_timeout* expires before every batch is submitted. In
the latter case the exception carries the partial
:class:`UpsertResponse` on its ``response`` attribute.
Note:
**Changed in 9.2.0.** Partial failures are aggregated rather than
raised, matching :meth:`upsert` with ``batch_size`` and the REST
client. Callers that relied on the raise should pass
``on_error="raise"``. The old raise discarded the partial count, so
no caller could tell what had landed; the new default reports it.
Because upserts are idempotent by vector ID, re-running the whole
DataFrame after a failure is also still safe.
Examples:
.. code-block:: python
import pandas as pd
from pinecone.grpc import GrpcIndex
idx = GrpcIndex(
host="article-search-abc123.svc.pinecone.io",
api_key="your-api-key",
)
df = pd.DataFrame([
{"id": "article-101", "values": [0.012, -0.087, 0.153]},
{"id": "article-102", "values": [0.045, 0.021, -0.064]},
])
response = idx.upsert_from_dataframe(df)
response.upserted_count
.. code-block:: python
df = pd.DataFrame([
{
"id": "article-101",
"values": [0.012, -0.087, 0.153],
"metadata": {"topic": "science", "year": 2024},
},
{
"id": "article-102",
"values": [0.045, 0.021, -0.064],
"metadata": {"topic": "technology", "year": 2024},
},
])
response = idx.upsert_from_dataframe(
df,
namespace="articles-en",
batch_size=100,
)
Give each batch a longer server-side deadline for large or slow
ingests:
.. code-block:: python
response = idx.upsert_from_dataframe(
df,
batch_size=200,
timeout=120.0,
)
"""
try:
import pandas as pd
except ImportError:
raise RuntimeError(
"pandas is required for upsert_from_dataframe, and is not a "
"dependency of this SDK — it is only needed by this one method. "
"Install it in your own environment: pip install pandas"
) from None
if not isinstance(df, pd.DataFrame):
raise PineconeValueError("df must be a pandas DataFrame")
validate_batch_size(batch_size)
resolved_on_error = _resolve_on_error(on_error)
resolved_concurrency = (
_default_max_concurrency() if max_concurrency is None else max_concurrency
)
require_in_range("max_concurrency", resolved_concurrency, 1, 64)
records: builtins.list[dict[str, Any]] = extract_records(df)
# Validate before submitting anything, so a malformed row cannot leave
# part of the frame ingested. VectorFactory would otherwise do this
# inside a worker thread, after earlier batches had already landed.
for record in records:
validate_vector_dict(record)
def _upsert_batch(batch: builtins.list[dict[str, Any]]) -> dict[str, Any]:
return self._channel.upsert(batch, namespace or None, timeout_s=timeout)
batch_result = batch_execute(
items=records,
operation=_upsert_batch,
batch_size=batch_size,
max_concurrency=resolved_concurrency,
show_progress=show_progress,
desc="Upserting",
executor=self._get_batch_executor(resolved_concurrency),
limiter_registry=self._limiter_registry,
host=self._limiter_host,
total_timeout=total_timeout,
)
response = _upsert_response_from(batch_result)
if batch_result.timed_out:
message = (
f"total_timeout of {total_timeout}s expired after "
f"{response.upserted_count} of {batch_result.total_item_count} vectors were "
f"upserted; retry the remainder with response.failed_items"
)
if resolved_on_error == "raise":
raise PineconeTimeoutError(message, response=response)
logger.warning(message)
return response
if batch_result.errors:
if resolved_on_error == "raise":
# All batches have settled by the time batch_execute returns, so
# nothing is left running server-side when this propagates.
error = min(batch_result.errors, key=lambda err: err.batch_index).error
error.response = response # type: ignore[attr-defined]
raise error
if on_error is None:
_warn_grpc_partial_failure_once(response)
return response
# ------------------------------------------------------------------
# Async (future-returning) variants
# ------------------------------------------------------------------
[docs]
def upsert_async(
self,
*,
vectors: Sequence[
Vector
| tuple[str, Sequence[float]]
| tuple[str, Sequence[float], Mapping[str, Any]]
| Mapping[str, Any]
],
namespace: str = "",
timeout: float | None = None,
) -> PineconeFuture[UpsertResponse]:
"""Submit an upsert operation and return a :class:`PineconeFuture`.
Same parameters as :meth:`upsert`, including ``timeout (float | None)``
which sets a per-call timeout in seconds.
Returns:
:class:`PineconeFuture` [:class:`UpsertResponse`] that resolves to
the upsert result.
Examples:
.. code-block:: python
future = index.upsert_async(
vectors=[("doc-42", [0.012, -0.087, 0.153])],
)
result = future.result()
result.upserted_count # 1
"""
future: PineconeFuture[UpsertResponse] = PineconeFuture(
self._executor.submit(
self.upsert, vectors=vectors, namespace=namespace, timeout=timeout
)
)
return future
[docs]
def query_async(
self,
*,
top_k: int,
vector: Sequence[float] | None = None,
id: str | None = None,
namespace: str = "",
filter: Mapping[str, Any] | None = None,
include_values: bool = False,
include_metadata: bool = False,
sparse_vector: SparseValues | Mapping[str, Any] | None = None,
scan_factor: float | None = None,
max_candidates: int | None = None,
timeout: float | None = None,
) -> PineconeFuture[QueryResponse]:
"""Submit a query operation and return a :class:`PineconeFuture`.
Same parameters as :meth:`query`, including ``timeout (float | None)``
which sets a per-call timeout in seconds.
Returns:
:class:`PineconeFuture` [:class:`QueryResponse`] that resolves to
the query result containing scored matches.
Examples:
.. code-block:: python
future = index.query_async(
vector=[0.012, -0.087, 0.153],
top_k=5,
)
result = future.result()
result.matches[0].id # 'doc-42'
result.matches[0].score # 0.95
"""
future: PineconeFuture[QueryResponse] = PineconeFuture(
self._executor.submit(
self.query,
top_k=top_k,
vector=vector,
id=id,
namespace=namespace,
filter=filter,
include_values=include_values,
include_metadata=include_metadata,
sparse_vector=sparse_vector,
scan_factor=scan_factor,
max_candidates=max_candidates,
timeout=timeout,
)
)
return future
[docs]
def fetch_async(
self,
*,
ids: Sequence[str],
namespace: str = "",
timeout: float | None = None,
) -> PineconeFuture[FetchResponse]:
"""Submit a fetch operation and return a :class:`PineconeFuture`.
Same parameters as :meth:`fetch`, including ``timeout (float | None)``
which sets a per-call timeout in seconds.
Returns:
:class:`PineconeFuture` [:class:`FetchResponse`] that resolves to
the fetched vectors keyed by ID.
Examples:
.. code-block:: python
future = index.fetch_async(ids=["doc-42", "doc-43"])
result = future.result()
result.vectors["doc-42"].values # [0.012, -0.087, 0.153]
"""
future: PineconeFuture[FetchResponse] = PineconeFuture(
self._executor.submit(self.fetch, ids=ids, namespace=namespace, timeout=timeout)
)
return future
[docs]
def delete_async(
self,
*,
ids: Sequence[str] | None = None,
delete_all: bool = False,
filter: Mapping[str, Any] | None = None,
namespace: str = "",
timeout: float | None = None,
) -> PineconeFuture[None]:
"""Submit a delete operation and return a :class:`PineconeFuture`.
Same parameters as :meth:`delete`, including ``timeout (float | None)``
which sets a per-call timeout in seconds.
Returns:
:class:`PineconeFuture` [None] that resolves when the delete
operation completes.
Examples:
.. code-block:: python
future = index.delete_async(ids=["doc-42", "doc-43"])
future.result()
.. code-block:: python
future = index.delete_async(delete_all=True, namespace="docs")
future.result()
"""
future: PineconeFuture[None] = PineconeFuture(
self._executor.submit(
self.delete,
ids=ids,
delete_all=delete_all,
filter=filter,
namespace=namespace,
timeout=timeout,
)
)
return future
[docs]
def update_async(
self,
*,
id: str | None = None,
values: Sequence[float] | None = None,
sparse_values: SparseValues | Mapping[str, Any] | None = None,
set_metadata: Mapping[str, Any] | None = None,
filter: Mapping[str, Any] | None = None,
namespace: str = "",
dry_run: bool = False,
timeout: float | None = None,
) -> PineconeFuture[UpdateResponse]:
"""Submit an update operation and return a :class:`PineconeFuture`.
Same parameters as :meth:`update`, including ``timeout (float | None)``
which sets a per-call timeout in seconds.
Returns:
:class:`PineconeFuture` [:class:`UpdateResponse`] that resolves to
the update result.
Examples:
.. code-block:: python
future = index.update_async(
id="article-101", values=[0.012, -0.087, 0.153]
)
result = future.result()
"""
return PineconeFuture(
self._executor.submit(
self.update,
id=id,
values=values,
sparse_values=sparse_values,
set_metadata=set_metadata,
filter=filter,
namespace=namespace,
dry_run=dry_run,
timeout=timeout,
)
)
[docs]
def query_namespaces_async(
self,
*,
vector: Sequence[float] | None = None,
namespaces: Sequence[str],
metric: str,
top_k: int | None = None,
filter: Mapping[str, Any] | None = None,
include_values: bool = False,
include_metadata: bool = False,
sparse_vector: SparseValues | Mapping[str, Any] | None = None,
scan_factor: float | None = None,
max_candidates: int | None = None,
timeout: float | None = None,
) -> PineconeFuture[QueryNamespacesResults]:
"""Submit a query_namespaces operation and return a :class:`PineconeFuture`.
Same parameters as :meth:`query_namespaces`, including ``timeout (float | None)``
which sets a per-call timeout in seconds.
Returns:
:class:`PineconeFuture` [:class:`QueryNamespacesResults`] that resolves to
the merged top-k matches across namespaces.
Examples:
.. code-block:: python
future = idx.query_namespaces_async(
vector=[0.012, -0.087, 0.153], # truncated; use your actual dimension
namespaces=["articles-en", "articles-fr", "articles-de"],
metric="cosine",
top_k=10,
)
results = future.result()
for match in results.matches:
print(match.id, match.score)
"""
return PineconeFuture(
self._executor.submit(
self.query_namespaces,
vector=vector,
namespaces=namespaces,
metric=metric,
top_k=top_k,
filter=filter,
include_values=include_values,
include_metadata=include_metadata,
sparse_vector=sparse_vector,
scan_factor=scan_factor,
max_candidates=max_candidates,
timeout=timeout,
)
)
[docs]
def upsert_records(
self,
*,
records: builtins.list[dict[str, Any]],
namespace: str,
timeout: float | None = None,
) -> UpsertRecordsResponse:
"""Upsert records for indexes with integrated inference.
Embeddings are generated server-side from the fields you provide, so
each record carries source data (e.g. text) rather than precomputed
vector values.
Args:
records: List of record dicts. Each must contain an ``_id`` or
``id`` field. Additional fields are passed through for
server-side embedding.
namespace (str): Target namespace (required). Unlike :meth:`upsert`,
namespace has no default because the records API requires an
explicit namespace (must be non-empty).
Returns:
:class:`UpsertRecordsResponse` with the count of records submitted.
Raises:
:exc:`PineconeValueError`: If namespace is not a string or is empty/whitespace,
records is empty, or a record is missing an identifier field.
:exc:`ApiError`: If the API returns an error response.
:exc:`PineconeConnectionError`: If a network-level connection
fails (DNS, refused, transport error).
:exc:`PineconeTimeoutError`: If the request does not complete
before *timeout* elapses.
Examples:
.. code-block:: python
pc = Pinecone(api_key="YOUR_API_KEY")
idx = pc.index("my-index", grpc=True)
response = idx.upsert_records(
namespace="articles-en",
records=[
{"_id": "article-101", "text": "Vector DBs enable similarity search."},
{"_id": "article-102", "text": "RAG combines search with LLMs."},
],
)
print(response.record_count)
"""
if not isinstance(namespace, str):
raise ValidationError("namespace must be a string")
if not namespace or not namespace.strip():
raise ValidationError("namespace must be a non-empty string")
if not records:
raise ValidationError("records must be a non-empty list")
for i, record in enumerate(records):
if "_id" not in record and "id" not in record:
raise ValidationError(f"Record at index {i} must contain an '_id' or 'id' field")
import orjson
normalized: builtins.list[dict[str, Any]] = []
for record in records:
r = dict(record)
if "_id" not in r and "id" in r:
r["_id"] = r.pop("id")
normalized.append(r)
ndjson_lines = [orjson.dumps(r).decode("utf-8") for r in normalized]
ndjson_body = "\n".join(ndjson_lines) + "\n"
logger.info(
"Upserting %d records into namespace %r (NDJSON via REST)", len(records), namespace
)
response = self._http.post(
f"/records/namespaces/{quote(namespace, safe='')}/upsert",
timeout=timeout,
content=ndjson_body.encode("utf-8"),
headers={"Content-Type": "application/x-ndjson"},
)
result = UpsertRecordsResponse(record_count=len(records))
result.response_info = extract_response_info(response)
return result
[docs]
def search(
self,
*,
namespace: str,
top_k: int | None = None,
inputs: SearchInputs | Mapping[str, Any] | None = None,
vector: Sequence[float] | Mapping[str, Any] | None = None,
id: str | None = None,
filter: Mapping[str, Any] | None = None,
fields: Sequence[str] | None = None,
rerank: RerankConfig | Mapping[str, Any] | None = None,
match_terms: Mapping[str, Any] | None = None,
query: SearchQuery | Mapping[str, Any] | None = None,
timeout: float | None = None,
) -> SearchRecordsResponse:
"""Search records by text, vector, or ID with optional reranking.
Delegates to the REST endpoint because the Pinecone gRPC API does not
expose a records search operation for integrated inference indexes.
.. note::
Use this method for indexes with integrated inference. For classic
indexes where you provide your own vectors, use :meth:`query`.
Args:
namespace (str): Namespace to search in (required).
top_k (int): Number of results to return (must be >= 1).
inputs (SearchInputs | dict[str, Any] | None): Inputs for
server-side embedding (e.g. ``{"text": "query text"}``).
vector (list[float] | None): Dense query vector values.
id (str | None): ID of an existing record to use as the query.
filter (dict[str, Any] | None): Metadata filter expression.
fields (list[str] | None): Field names to include in results.
When ``None``, the server returns all available fields.
rerank (RerankConfig | dict[str, Any] | None): Reranking
configuration with ``model`` (required), ``rank_fields``
(required), and optional ``top_n``, ``parameters``, ``query``
keys. Use :class:`RerankConfig` for IDE autocompletion.
match_terms (dict[str, Any] | None): Term-matching constraint for
sparse search. Requires keys ``"strategy"`` (currently only
``"all"``) and ``"terms"`` (list of strings).
Valid only on a text query — combined with ``vector`` or ``id``
it is rejected — and only on a sparse index whose embedding model
supports it; the server names the supported model when it
refuses. ``None`` disables term matching.
query (dict[str, Any] | None): Legacy query body containing
``top_k`` plus one of ``inputs``, ``vector``, or ``id``. Prefer
passing these fields directly.
Returns:
:class:`SearchRecordsResponse` with hits and usage statistics.
Raises:
:exc:`PineconeValueError`: If ``namespace`` is not a string, ``top_k < 1``,
or ``rerank`` is missing required keys.
:exc:`ApiError`: If the API returns an error response.
:exc:`PineconeConnectionError`: If a network-level connection
fails (DNS, refused, transport error).
:exc:`PineconeTimeoutError`: If the request does not complete before
*timeout* elapses.
Examples:
.. code-block:: python
response = idx.search(
namespace="articles-en",
top_k=10,
inputs={"text": "benefits of vector databases for search"},
)
for hit in response.result.hits:
print(hit.id, hit.score)
Search with reranking:
.. code-block:: python
response = idx.search(
namespace="articles-en",
top_k=10,
inputs={"text": "benefits of vector databases"},
rerank={
"model": "bge-reranker-v2-m3",
"rank_fields": ["text"],
"top_n": 5,
},
)
for hit in response.result.hits:
print(hit.id, hit.score)
.. note::
Use inline ``rerank`` when searching and reranking in a single call.
Use ``pc.inference.rerank()`` when reranking results from a different
source or when you need to rerank without searching.
"""
if not isinstance(namespace, str):
raise ValidationError("namespace must be a string")
if not namespace or not namespace.strip():
raise ValidationError("namespace must be a non-empty string")
body = _build_search_records_body(
method_name="GrpcIndex.search",
top_k=top_k,
inputs=inputs,
vector=vector,
id=id,
filter=filter,
fields=fields,
rerank=rerank,
match_terms=match_terms,
query=query,
)
logger.info(
"Searching namespace %r with top_k=%d (via REST)",
namespace,
body["query"]["top_k"],
)
response = self._http.post(
f"/records/namespaces/{quote(namespace, safe='')}/search", timeout=timeout, json=body
)
result = self._adapter.to_search_response(response.content)
result.response_info = extract_response_info(response)
return result
[docs]
def search_records(
self,
*,
namespace: str,
top_k: int | None = None,
inputs: SearchInputs | Mapping[str, Any] | None = None,
vector: Sequence[float] | Mapping[str, Any] | None = None,
id: str | None = None,
filter: Mapping[str, Any] | None = None,
fields: Sequence[str] | None = None,
rerank: RerankConfig | Mapping[str, Any] | None = None,
match_terms: Mapping[str, Any] | None = None,
query: SearchQuery | Mapping[str, Any] | None = None,
timeout: float | None = None,
) -> SearchRecordsResponse:
"""Alias for :meth:`search`, kept for backwards compatibility.
Prefer calling :meth:`search` directly.
Examples:
.. code-block:: python
response = idx.search_records(
namespace="articles-en",
top_k=10,
inputs={"text": "benefits of vector databases for search"},
)
"""
return self.search(
namespace=namespace,
top_k=top_k,
inputs=inputs,
vector=vector,
id=id,
filter=filter,
fields=fields,
rerank=rerank,
match_terms=match_terms,
query=query,
timeout=timeout,
)
[docs]
def list_namespaces_paginated(
self,
*,
prefix: str | None = None,
limit: int | None = None,
pagination_token: str | None = None,
timeout: float | None = None,
) -> ListNamespacesResponse:
"""Fetch a single page of namespace descriptions.
Args:
prefix (str | None): Return only namespaces whose names start with this
prefix. Must be ASCII, must not contain the NUL character, and must
be at most 512 characters. The empty prefix matches every namespace.
limit (int | None): Maximum number of namespaces to return in this page,
1-100.
pagination_token (str | None): Token from a previous response to fetch the next page.
timeout (float | None): Per-call timeout in seconds.
Returns:
:class:`ListNamespacesResponse` with namespace descriptions, pagination info,
and total count. Each description carries ``size_bytes``.
Raises:
:exc:`PineconeValueError`: If *prefix* or *limit* violates the rules
above. Raised locally, before the request is sent, with the same
message the REST and asyncio clients raise.
:exc:`PineconeTimeoutError`: If the request does not complete before
*timeout* elapses.
Examples:
.. code-block:: python
page = idx.list_namespaces_paginated(prefix="prod-", limit=50)
for ns in page.namespaces:
print(ns.name, ns.record_count, ns.size_bytes)
"""
if prefix is not None:
require_valid_namespace_prefix("prefix", prefix)
if limit is not None:
require_valid_namespace_limit("limit", limit)
logger.info("Listing namespaces (paginated) via gRPC")
result = self._channel.list_namespaces(
prefix=prefix,
limit=limit,
pagination_token=pagination_token,
timeout_s=timeout,
)
namespaces = [
_dict_to_namespace_description(ns_data) for ns_data in result.get("namespaces", [])
]
pagination: Pagination | None = None
raw_pag = result.get("pagination")
if raw_pag is not None:
pagination = Pagination(next=raw_pag.get("next"))
return ListNamespacesResponse(
namespaces=namespaces,
pagination=pagination,
total_count=result.get("total_count", 0),
)
[docs]
def list_namespaces(
self,
*,
prefix: str | None = None,
limit: int | None = None,
timeout: float | None = None,
) -> Iterator[ListNamespacesResponse]:
"""List namespaces, automatically following pagination.
Yields one :class:`ListNamespacesResponse` per page. The generator
automatically follows pagination tokens until all pages have been
retrieved.
Args:
prefix (str | None): Return only namespaces whose names start with this
prefix. Must be ASCII, must not contain the NUL character, and must
be at most 512 characters. The empty prefix matches every namespace.
limit (int | None): Maximum number of namespaces to return per page, 1-100.
timeout (float | None): Per-call timeout in seconds.
Yields:
:class:`ListNamespacesResponse` for each page of results. Each
:class:`NamespaceDescription` carries ``size_bytes``.
Raises:
:exc:`PineconeValueError`: If *prefix* or *limit* violates the rules
above. Raised on the first iteration, before the request is sent.
:exc:`PineconeTimeoutError`: If a page request does not complete
before *timeout* elapses.
Examples:
.. code-block:: python
for page in idx.list_namespaces(prefix="prod-"):
for ns in page.namespaces:
print(ns.name, ns.record_count, ns.size_bytes)
"""
pagination_token: str | None = None
while True:
page = self.list_namespaces_paginated(
prefix=prefix,
limit=limit,
pagination_token=pagination_token,
timeout=timeout,
)
if page.namespaces:
yield page
if page.pagination is not None and page.pagination.next is not None:
pagination_token = page.pagination.next
else:
break
[docs]
def create_namespace(
self,
*,
name: str,
schema: dict[str, Any] | None = None,
timeout: float | None = None,
) -> NamespaceDescription:
"""Create a named namespace in the index.
Args:
name (str): Name for the new namespace. Must be ASCII, must not
contain the NUL character, and must be 1-512 characters long.
``__default__`` is reserved and cannot be created: it names the
namespace requests address when they omit a namespace, so it
always exists.
schema (dict[str, Any] | None): Optional metadata-index configuration,
``{"fields": {<field>: {"filterable": True}}}``. Omitting it does
not mean "index everything": the namespace inherits the index's
own metadata-index configuration, so an index that restricts which
fields are indexed passes that restriction on. Supply *schema* to
override the inherited configuration for this namespace, indexing
exactly the fields listed. ``filterable`` is required on each field
and must be ``True`` — to leave a field unindexed, omit it from
``fields``.
timeout (float | None): Per-call timeout in seconds.
Returns:
:class:`NamespaceDescription` with the namespace name, record count,
schema, indexed fields, and ``size_bytes``.
Raises:
:exc:`PineconeValueError`: If *name* violates the rules above, or
*schema* is malformed. Raised locally, before the request is
sent, with the same message the REST and asyncio clients raise.
:exc:`PineconeTimeoutError`: If the request does not complete before
*timeout* elapses.
Examples:
.. code-block:: python
ns = idx.create_namespace(name="movies-en")
print(ns.name, ns.record_count, ns.size_bytes)
"""
require_creatable_namespace_name("name", name)
if schema is not None:
require_valid_namespace_schema("schema", schema)
logger.info("Creating namespace %r via gRPC", name)
result = self._channel.create_namespace(name, schema, timeout_s=timeout)
return _dict_to_namespace_description(result)
[docs]
def describe_namespace(
self,
*,
name: str | None = None,
timeout: float | None = None,
**kwargs: str,
) -> NamespaceDescription:
"""Describe a namespace by name.
This operation is rate limited per index, independently of the other
namespace operations. Prefer :meth:`list_namespaces` when describing more
than one namespace: it returns the same information for every namespace
in a single request and is not subject to that limit.
Args:
name (str): Name of the namespace to describe. Must be ASCII, must not
contain the NUL character, and must be 1-512 characters long.
``__default__`` is accepted and describes the namespace requests
address when they omit one.
timeout (float | None): Per-call timeout in seconds.
Returns:
:class:`NamespaceDescription` with the namespace name, record count,
schema, indexed fields, and ``size_bytes``.
Raises:
:exc:`PineconeValueError`: If *name* violates the rules above.
Raised locally, before the request is sent, with the same
message the REST and asyncio clients raise.
:exc:`TypeError`: If unexpected keyword arguments are passed.
:exc:`PineconeTimeoutError`: If the request does not complete before
*timeout* elapses.
Examples:
.. code-block:: python
ns = idx.describe_namespace(name="movies-en")
print(ns.name, ns.record_count, ns.size_bytes)
"""
legacy_namespace: str | None = kwargs.pop("namespace", None)
if kwargs:
raise TypeError(
f"describe_namespace() got unexpected keyword arguments: {sorted(kwargs)!r}"
)
if name is not None and legacy_namespace is not None:
raise ValidationError("Provide either name= or namespace=, not both")
effective: str = name if name is not None else (legacy_namespace or "")
require_valid_namespace_name("name", effective)
logger.info("Describing namespace %r via gRPC", effective)
result = self._channel.describe_namespace(effective, timeout_s=timeout)
return _dict_to_namespace_description(result)
[docs]
def delete_namespace(
self,
*,
name: str | None = None,
timeout: float | None = None,
**kwargs: str,
) -> None:
"""Delete a namespace by name, removing all its vectors.
Args:
name (str): Name of the namespace to delete. Must be ASCII, must not
contain the NUL character, and must be 1-512 characters long.
timeout (float | None): Per-call timeout in seconds.
Returns:
None — a successful delete returns no payload.
Raises:
:exc:`PineconeValueError`: If *name* violates the rules above.
Raised locally, before the request is sent, with the same
message the REST and asyncio clients raise.
:exc:`TypeError`: If unexpected keyword arguments are passed.
:exc:`PineconeTimeoutError`: If the request does not complete before
*timeout* elapses.
Examples:
.. code-block:: python
idx.delete_namespace(name="movies-en")
"""
legacy_namespace: str | None = kwargs.pop("namespace", None)
if kwargs:
raise TypeError(
f"delete_namespace() got unexpected keyword arguments: {sorted(kwargs)!r}"
)
if name is not None and legacy_namespace is not None:
raise ValidationError("Provide either name= or namespace=, not both")
effective: str = name if name is not None else (legacy_namespace or "")
require_valid_namespace_name("name", effective)
logger.info("Deleting namespace %r via gRPC", effective)
self._channel.delete_namespace(effective, timeout_s=timeout)
def _validate_import_id(self, id: str | int) -> str:
"""Validate and normalize an import operation ID.
Args:
id: Import operation ID. If int, converted to str silently.
Returns:
The validated string ID.
Raises:
:exc:`PineconeValueError`: If the ID is empty or exceeds 1000 characters.
"""
str_id = str(id) if isinstance(id, int) else id
if not str_id or len(str_id) > 1000:
raise ValidationError(
"import id must be between 1 and 1000 characters, "
f"got {len(str_id) if str_id else 0}"
)
return str_id
[docs]
def start_import(
self,
uri: str,
*,
error_mode: str | None = None,
integration_id: str | None = None,
) -> StartImportResponse:
"""Start a bulk import operation from an external data source.
Initiates an asynchronous bulk import of vectors from cloud storage
into the index. The import runs server-side; use :meth:`describe_import`
to poll for progress and completion.
.. note::
The import URI must point to a directory of Parquet files in cloud
storage. Each Parquet file must follow the Pinecone-required schema.
See
`Pinecone import docs <https://docs.pinecone.io/guides/data/understanding-imports>`_
for the required Parquet schema and supported storage formats.
Args:
uri (str): Directory prefix holding the Parquet files, not a single
file. Three forms are accepted: ``s3://`` for Amazon S3,
``gs://`` for Google Cloud Storage, and an ``https://`` URL
naming an Azure Blob Storage container. ``s3://`` additionally
requires that the index itself be hosted on AWS.
error_mode (str | None): How to handle a record the import cannot
read. ``"continue"`` skips it and imports the rest; ``"abort"``
ends the whole import at the first such record. Case-insensitive.
Defaults to ``"abort"`` when omitted, so an unreadable record
fails the import unless you opt into skipping.
integration_id (str | None): Optional integration ID for the import.
Returns:
:class:`StartImportResponse` with the ID of the created import
operation.
Raises:
:exc:`PineconeValueError`: If ``error_mode`` is supplied but not
``"continue"`` or ``"abort"``.
:exc:`ApiError`: If ``uri`` is empty or longer than the server
accepts, uses an unsupported scheme, is an ``s3://`` URI on an
index not hosted on AWS, or names an S3 directory bucket, which
imports do not support, or if the API otherwise returns an
error response.
:exc:`PineconeConnectionError`: If a network-level connection
fails (DNS, refused, transport error).
:exc:`PineconeTimeoutError`: If the request does not complete
before the configured timeout elapses.
Examples:
.. code-block:: python
# Start an import and poll until complete
import time
response = idx.start_import(uri="s3://my-bucket/vectors/")
import_id = response.id
# Poll until the import finishes
import_op = idx.describe_import(import_id)
while import_op.status not in ("Completed", "Failed", "Cancelled"):
time.sleep(10)
import_op = idx.describe_import(import_id)
print(f"Status: {import_op.status}, records imported: {import_op.records_imported}")
# Skip unreadable records instead of failing the import
response = idx.start_import(
uri="s3://my-bucket/vectors/",
error_mode="continue",
)
.. seealso::
- :meth:`upsert` — for upserting vectors directly in small
batches (single request per call).
- :meth:`upsert_records` — for indexes with integrated inference
(text in, server-side embedding).
- :meth:`upsert_from_dataframe` — for loading vectors from a
pandas DataFrame with automatic batching.
"""
if error_mode is not None:
error_mode = error_mode.lower()
if error_mode not in ("continue", "abort"):
raise ValidationError(
f"error_mode must be 'continue' or 'abort', got {error_mode!r}"
)
body: dict[str, Any] = {"uri": uri}
if error_mode is not None:
body["errorMode"] = {"onError": error_mode}
if integration_id is not None:
body["integrationId"] = integration_id
logger.info("Starting bulk import from %s", uri)
response = self._http.post("/bulk/imports", json=body)
return self._imports_adapter.to_start_import_response(response.content)
[docs]
def describe_import(self, id: str | int) -> ImportModel:
"""Describe a bulk import operation by ID.
Args:
id: Import operation ID. Integers are converted to strings silently.
Returns:
:class:`ImportModel` with the import operation details.
Raises:
:exc:`PineconeValueError`: If the ID is empty or exceeds 1000 characters.
:exc:`ApiError`: If the API returns an error response.
:exc:`PineconeConnectionError`: If a network-level connection
fails (DNS, refused, transport error).
:exc:`PineconeTimeoutError`: If the request does not complete
before the configured timeout elapses.
Examples:
.. code-block:: python
import_op = idx.describe_import("import-123")
print(import_op.status, import_op.percent_complete)
"""
str_id = self._validate_import_id(id)
logger.info("Describing import %s", str_id)
response = self._http.get(f"/bulk/imports/{quote(str_id, safe='')}")
return self._imports_adapter.to_import_model(response.content)
[docs]
def cancel_import(self, id: str | int) -> None:
"""Cancel a running bulk import operation by ID.
Args:
id (str | int): ID of the import to cancel, as returned by
:meth:`start_import`. Integers are converted to strings silently.
Returns:
None — a successful cancellation returns no payload.
Raises:
:exc:`PineconeValueError`: If the ID is empty or exceeds 1000 characters.
:exc:`ApiError`: If the API returns an error response.
:exc:`PineconeConnectionError`: If a network-level connection
fails (DNS, refused, transport error).
:exc:`PineconeTimeoutError`: If the request does not complete
before the configured timeout elapses.
Examples:
.. code-block:: python
idx.cancel_import("import-123")
"""
str_id = self._validate_import_id(id)
logger.info("Cancelling import %s", str_id)
self._http.delete(f"/bulk/imports/{quote(str_id, safe='')}")
[docs]
def list_imports(
self,
*,
limit: int | None = None,
pagination_token: str | None = None,
) -> Iterator[ImportModel]:
"""List bulk import operations, automatically following pagination.
Yields individual :class:`ImportModel` objects, fetching additional
pages transparently until all results have been returned. Prefer
:meth:`list_imports_paginated` to control pagination yourself.
Args:
limit (int | None): Maximum number of imports per page. Omit to let
the server choose the page size.
pagination_token (str | None): Token to resume pagination
from a previous call.
Yields:
:class:`ImportModel` for each import operation.
Raises:
:exc:`ApiError`: If the API returns an error response.
:exc:`PineconeConnectionError`: If a network-level connection
fails (DNS, refused, transport error).
:exc:`PineconeTimeoutError`: If a page request does not complete
before the configured timeout elapses.
Examples:
.. code-block:: python
for imp in idx.list_imports():
print(imp.id, imp.status)
"""
params: dict[str, Any] = {}
if limit is not None:
params["limit"] = limit
if pagination_token is not None:
params["paginationToken"] = pagination_token
while True:
response = self._http.get("/bulk/imports", params=params)
import_list = self._imports_adapter.to_import_list(response.content)
yield from import_list
next_token = import_list.pagination.next if import_list.pagination else None
if next_token is None:
break
params["paginationToken"] = next_token
[docs]
def list_imports_paginated(
self,
*,
limit: int | None = None,
pagination_token: str | None = None,
) -> ImportList:
"""Fetch a single page of bulk import operations.
Returns an :class:`ImportList` for one page. The caller is responsible
for managing the pagination token. Prefer :meth:`list_imports` to have
pagination handled automatically.
Args:
limit (int | None): Maximum number of imports to return in this page.
pagination_token (str | None): Token from a previous response to
fetch the next page.
Returns:
:class:`ImportList` for the requested page, iterable over its
:class:`ImportModel` entries. Its ``pagination.next`` field holds
the token for the next page, or ``None`` once there are no more.
Raises:
:exc:`ApiError`: If the API returns an error response.
:exc:`PineconeConnectionError`: If a network-level connection
fails (DNS, refused, transport error).
:exc:`PineconeTimeoutError`: If the request does not complete
before the configured timeout elapses.
Examples:
.. code-block:: python
page = idx.list_imports_paginated(limit=10)
for imp in page:
print(imp.id, imp.status)
next_token = page.pagination.next if page.pagination else None
"""
params: dict[str, Any] = {}
if limit is not None:
params["limit"] = limit
if pagination_token is not None:
params["paginationToken"] = pagination_token
response = self._http.get("/bulk/imports", params=params)
return self._imports_adapter.to_import_list(response.content)
[docs]
def close(self) -> None:
"""Close the connection to the index and release background resources.
Waits for any in-flight ``*_async`` submissions to finish, then shuts
down the worker pools used for batch upserts and closes the network
connection. Call this when you are done issuing requests through this
client and are not using it as a context manager.
Examples:
.. code-block:: python
idx = pc.index("my-index", grpc=True)
idx.upsert(vectors=[...])
idx.close()
"""
self._executor.shutdown(wait=True)
with self._batch_executor_lock:
executors = list(self._batch_executors.values())
self._batch_executors.clear()
for executor in executors:
executor.shutdown(wait=False)
self._http.close()
if hasattr(self._channel, "close"):
self._channel.close()
[docs]
def __enter__(self) -> GrpcIndex:
"""Enter a context manager block, returning this client unchanged.
Examples:
.. code-block:: python
with pc.index("my-index", grpc=True) as idx:
idx.upsert(vectors=[...])
"""
return self
[docs]
def __exit__(self, *args: Any) -> None:
"""Exit the context manager block, calling :meth:`close`."""
self.close()
# Legacy capitalisation alias (BCG-141).
GRPCIndex = GrpcIndex
# Legacy name (renamed from PineconeGrpcFuture in the rewrite — BCG-143).
PineconeGrpcFuture = PineconeFuture
from pinecone.grpc.pinecone_grpc import PineconeGRPC # noqa: E402
__all__ = ["GRPCIndex", "GrpcIndex", "PineconeGRPC", "PineconeGrpcFuture"]