Pinecone Python SDK

Some SDK operations return results in pages rather than a single list. To handle this, the SDK provides Paginator (for synchronous code) and AsyncPaginator (for asynchronous code). These objects allow you to iterate over results lazily, fetching subsequent pages only when requested.

Both paginators support the following interface:

• __iter__ / __aiter__: Iterate over individual items across all pages sequentially.
• .pages(): Iterate over Page objects instead of individual items.
• .to_list(): Fetch all items from all pages into a single list in one call.
• .pagination_token: Access the token for the next page; returns None when all pages have been consumed.

# Sync example: iterating over items
from pinecone import Pinecone

pc = Pinecone()
for assistant in pc.assistants.list():
    print(assistant.name)

# Async example: iterating over items
import asyncio
from pinecone import AsyncPinecone

async def main() -> None:
    async with AsyncPinecone() as pc:
        async for assistant in pc.assistants.list():
            print(assistant.name)

asyncio.run(main())

When performing bulk operations like upsert(), the SDK uses an AIMD (Additive Increase, Multiplicative Decrease) algorithm to manage concurrency.

• Throttling: If the SDK receives a retryable response (like 429 or 503), it halves the effective concurrency floor for that host.
• Recovery: After a streak of successful requests, it recovers by one slot.
• Configuration: You do not configure the limiter directly. The max_concurrency parameter in methods like upsert() acts as a ceiling. The SDK self-tunes the actual number of in-flight requests between 1 and that ceiling based on server capacity.

from pinecone import Pinecone

pc = Pinecone()
index = pc.index(host="product-search-abc123.svc.pinecone.io")

# max_concurrency=8 is the ceiling. The SDK will scale below this if throttled.
response = index.upsert(
    vectors=large_list,
    batch_size=200,
    max_concurrency=8,
)

The pc.inference.embed method returns an EmbeddingsList object. This object contains:

• .data: A list of DenseEmbedding or SparseEmbedding objects (one per input).
• .model: The name of the model used.
• .usage.total_tokens: The total token count consumed by the request.

Accessing Dense Embeddings: Iterate through the result and access .values to get a list of floats.

Accessing Sparse Embeddings: For models like pinecone-sparse-english-v0, access .sparse_indices and .sparse_values from the embedding object.

Hybrid Embeddings: Note that some models return hybrid (dense + sparse) embeddings as two separate items per input.

# For Dense Embeddings
for emb in result:
    print(emb.values)       # DenseEmbedding: list of floats

# For Sparse Embeddings
result = pc.inference.embed(
    model="pinecone-sparse-english-v0",
    inputs=["machine learning frameworks"],
)
sparse = result.data[0]
print(sparse.sparse_indices)
print(sparse.sparse_values)

All exceptions raised by the SDK are subclasses of PineconeError. You can catch all SDK-related errors using a single except PineconeError block, or catch specific subclasses to handle different failure modes (like network issues vs. API errors) uniquely.

Hierarchy Overview:

• PineconeError (Base class)
  • ApiError (HTTP error responses from the server)
    • NotFoundError (404)
    • ConflictError (409)
    • UnauthorizedError (401)
    • ForbiddenError (403)
    • RateLimitError (429)
    • ServiceError (5xx)
  • PineconeConnectionError (Network-level failures like DNS or refused connections)
  • PineconeTimeoutError (Operation exceeded its timeout)
  • PineconeValueError (Invalid value passed to the SDK)
  • PineconeTypeError (Wrong type passed to the SDK)

from pinecone.errors import PineconeError

try:
    # SDK operation
    pass
except PineconeError as e:
    print(f"An SDK error occurred: {e}")

The AsyncPinecone class is the asynchronous control-plane client. It is designed to be used within an async with block to manage high-level resources like indexes, collections, backups, and assistants.

Sub-clients for specific resource types are accessed as properties (e.g., pc.indexes, pc.collections, pc.backups, pc.restore_jobs, pc.inference, pc.assistants) and are lazily initialized upon their first access.

Important Note on Index Access: Unlike the synchronous Pinecone client, AsyncPinecone.index() cannot automatically resolve an index host using only its name. To connect to an index, you must either:

1. Call await pc.indexes.describe("index-name") first to populate the internal cache, then call pc.index("index-name").
2. Explicitly provide the host by calling desc = await pc.indexes.describe("index-name") and then passing host=desc.host to pc.index().

from pinecone import AsyncPinecone

async with AsyncPinecone(api_key="your-api-key") as pc:
    # 1. Describe the index to get host information
    desc = await pc.indexes.describe("my-index")
    
    # 2. Connect to the index using the host
    index = pc.index(host=desc.host)
    
    # 3. Use the index within an async context manager
    async with index:
        results = await index.query(
            vector=[0.012, -0.087, 0.153],
            top_k=10,
        )

The behavior of index.upsert() changed when using the batch_size parameter.

v8 Behavior: Raised an exception on the first batch failure, potentially leaving subsequent batches unattempted. v9 Behavior: Batches are submitted concurrently. Failures that exceed the retry budget are captured in the returned UpsertResponse rather than being raised as exceptions.

How to handle errors in v9: Instead of using try/except to catch batch failures, inspect the has_errors property on the returned response object.

Example:

# v9 — inspect response.has_errors instead
response = idx.upsert(vectors=batch, batch_size=100)
if response.has_errors:
    # Optionally retry only the failures:
    idx.upsert(vectors=response.failed_items, batch_size=100)
    # ...or roll back if any failure is unacceptable

Note: Single-request upserts (batch_size=None, the default) retain the v8 'raise-on-failure' semantics.

# v9 — inspect response.has_errors instead
response = idx.upsert(vectors=batch, batch_size=100)
if response.has_errors:
    # Optionally retry only the failures:
    idx.upsert(vectors=response.failed_items, batch_size=100)
    # …or roll back if any failure is unacceptable