Source code for pinecone.async_client.documents

"""AsyncDocuments namespace — document data-plane operations (2026-07 API)."""

from __future__ import annotations

import logging
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any

from pinecone._internal.adapters.documents_adapter import DocumentsAdapter
from pinecone._internal.batch import async_batch_execute
from pinecone._internal.documents_helpers import (
    _build_delete_documents_body,
    _build_fetch_documents_body,
    _build_list_documents_body,
    _build_search_documents_body,
    _build_update_documents_body,
    _encode_document_namespace,
    _validate_documents,
)
from pinecone._internal.keyword_only import keyword_only_methods
from pinecone._internal.validation import require_in_range
from pinecone.models.batch import BatchResult
from pinecone.models.documents.document import DocumentRecord, UpdateDocumentRecord
from pinecone.models.documents.responses import (
    DeleteDocumentsResponse,
    FetchDocumentsResponse,
    ListedDocumentRecord,
    SearchDocumentsResponse,
    UpdateDocumentsResponse,
    UpsertDocumentsResponse,
)
from pinecone.models.documents.score_by import DocumentScoringMethod
from pinecone.models.pagination import AsyncPaginator, Page

if TYPE_CHECKING:
    from pinecone._internal.http_client import AsyncHTTPClient

logger = logging.getLogger(__name__)


[docs] @keyword_only_methods class AsyncDocuments: """Document data-plane operations for a schema-based index (2026-07 API). Accessed via :attr:`~pinecone.async_client.async_index.AsyncIndex.documents`. Not constructed directly — the parent :class:`~pinecone.async_client.async_index.AsyncIndex` builds and caches its own instance on first access. Examples: .. code-block:: python from pinecone import AsyncPinecone pc = AsyncPinecone(api_key="your-api-key") index = await pc.index(name="articles-en") async with index: await index.documents.upsert( namespace="articles-en", documents=[{"_id": "article-101", "title": "Intro to vectors"}], ) """
[docs] def __init__(self, *, http: AsyncHTTPClient) -> None: self._http = http
def __repr__(self) -> str: return "AsyncDocuments()"
[docs] async def upsert( self, *, namespace: str, documents: Sequence[Mapping[str, Any] | DocumentRecord], timeout: float | None = None, ) -> UpsertDocumentsResponse: """Upsert documents into a namespace. Each document must include an ``_id`` field (a unique, non-empty ASCII string of at most 512 characters) along with fields defined in the index schema or arbitrary metadata fields. If a document with the same ``_id`` already exists in the namespace, it is overwritten. Pinecone applies the upsert asynchronously, so documents may not be immediately visible to :meth:`search` or :meth:`fetch`. Args: namespace (str): Target namespace (required, non-empty). documents: The documents to upsert (1-1000 per request). Each element is a dict with an ``_id`` key or a :class:`~pinecone.models.documents.document.DocumentRecord`. For larger lists, use :meth:`batch_upsert`. timeout (float | None): Per-request timeout in seconds. Overrides the client-level default for this call only. Returns: :class:`UpsertDocumentsResponse` with the count of documents accepted for upsert. Raises: :exc:`PineconeValueError`: If ``namespace`` is empty, ``documents`` is empty or over 1000 entries, or any document has a missing, empty, non-string, non-ASCII, over-512-character, or duplicate ``_id`` — the message names the offending document's position. :exc:`ApiError`: If one request exceeds the server's cap on encoded request size — a document count inside the accepted range can still be too large. Send fewer documents per request, or use :meth:`batch_upsert`. :exc:`ApiError`: If the API returns an error response; the server's error text is surfaced intact. :exc:`PineconeConnectionError`: If a network-level connection fails (DNS, refused, transport error). :exc:`PineconeTimeoutError`: If the request exceeds the configured timeout. Examples: .. code-block:: python response = await idx.documents.upsert( namespace="articles-en", documents=[ {"_id": "article-101", "title": "Intro to vectors"}, {"_id": "article-102", "title": "Advanced retrieval"}, ], ) print(response.upserted_count) .. seealso:: - :meth:`batch_upsert` — for upserting large document lists in parallel batches. - :meth:`~pinecone.async_client.async_index.AsyncIndex.upsert` — for indexes where you provide your own vectors. """ segment = _encode_document_namespace(namespace) normalized = _validate_documents(documents) logger.info("Upserting %d documents into namespace %r", len(normalized), namespace) response = await self._http.post( f"/namespaces/{segment}/documents/upsert", timeout=timeout, json={"documents": normalized}, ) return DocumentsAdapter.to_upsert_response(response)
[docs] async def batch_upsert( self, *, namespace: str, documents: Sequence[Mapping[str, Any] | DocumentRecord], batch_size: int = 50, max_concurrency: int = 4, show_progress: bool = True, timeout: float | None = None, ) -> BatchResult: """Upsert a large list of documents in parallel batches. Splits *documents* into chunks of *batch_size* and submits them concurrently behind an ``asyncio.Semaphore`` of *max_concurrency* slots. Per-batch HTTP failures are captured in the returned :class:`~pinecone.models.batch.BatchResult` rather than raised, so one failed batch does not abort the rest; retry only the failures by passing ``result.failed_items`` back in. Args: namespace (str): Target namespace (required, non-empty). documents: Documents to upsert. Each element is a dict with an ``_id`` key or a :class:`~pinecone.models.documents.document.DocumentRecord`; IDs must be unique across the whole list. batch_size (int): Maximum documents per request (1-1000, default 50). max_concurrency (int): Asyncio concurrency limit for concurrent batch requests (1-64, default 4). show_progress (bool): Display a progress bar when ``tqdm`` is installed. Defaults to ``True``. timeout (float | None): Per-request timeout in seconds applied to each batch's request — not to the whole call. Returns: :class:`~pinecone.models.batch.BatchResult` with aggregated success and failure counts; per-batch errors are in ``result.errors`` and the affected documents in ``result.failed_items``. Raises: :exc:`PineconeValueError`: If ``namespace`` is empty, ``documents`` is empty or contains an invalid or duplicate ``_id``, ``batch_size`` is outside [1, 1000], or ``max_concurrency`` is outside [1, 64]. Examples: .. code-block:: python documents = [ {"_id": f"article-{i}", "title": f"Article {i}"} for i in range(5000) ] result = await idx.documents.batch_upsert( namespace="articles-en", documents=documents, batch_size=100, max_concurrency=8, ) print(result.successful_item_count, result.failed_item_count) .. seealso:: - :meth:`upsert` — for a single-request upsert of up to 1000 documents. """ require_in_range("batch_size", batch_size, 1, 1000) require_in_range("max_concurrency", max_concurrency, 1, 64) segment = _encode_document_namespace(namespace) normalized = _validate_documents(documents, max_documents=None) async def _operation(chunk: list[dict[str, Any]]) -> UpsertDocumentsResponse: response = await self._http.post( f"/namespaces/{segment}/documents/upsert", timeout=timeout, json={"documents": chunk}, ) return DocumentsAdapter.to_upsert_response(response) return await async_batch_execute( items=normalized, operation=_operation, batch_size=batch_size, max_concurrency=max_concurrency, show_progress=show_progress, desc="Upserting documents", )
[docs] async def search( self, *, namespace: str, score_by: Sequence[DocumentScoringMethod | Mapping[str, Any]], top_k: int, include_fields: Sequence[str] | None = None, filter: Mapping[str, Any] | None = None, timeout: float | None = None, ) -> SearchDocumentsResponse: """Search documents in a namespace using one or more scoring methods. Returns the ``top_k`` most similar documents ranked by the given scoring methods (dense vector, sparse vector, BM25 text, or Lucene query string similarity). Args: namespace (str): Namespace to search (required, non-empty). score_by: Scoring methods to rank documents by (1-100 clauses). Items are typed variants (:class:`~pinecone.models.documents.score_by.TextQuery`, :class:`~pinecone.models.documents.score_by.QueryStringQuery`, :class:`~pinecone.models.documents.score_by.DenseVectorQuery`, :class:`~pinecone.models.documents.score_by.SparseVectorQuery`) or plain dicts with a ``type`` key. ``text`` and ``query_string`` clauses may be combined; a ``dense_vector`` or ``sparse_vector`` clause must appear alone. top_k (int): Number of top-ranked documents to return (1-10000). include_fields: Document fields to include in the results. ``None`` (default) returns all fields; ``[]`` returns only ``_id`` and score. filter: Metadata filter expression restricting the documents searched, or ``None``. timeout (float | None): Per-request timeout in seconds. Overrides the client-level default for this call only. Returns: :class:`SearchDocumentsResponse` with ``matches`` (ordered from most to least similar), ``namespace``, and ``usage``. Raises: :exc:`PineconeValueError`: If ``namespace`` is empty, ``score_by`` is empty, over 100 clauses, or combines a vector clause with any other clause, or ``top_k`` is outside [1, 10000]. :exc:`ApiError`: If the API returns an error response; the server's error text is surfaced intact. :exc:`PineconeConnectionError`: If a network-level connection fails (DNS, refused, transport error). :exc:`PineconeTimeoutError`: If the request exceeds the configured timeout. Examples: .. code-block:: python from pinecone import TextQuery response = await idx.documents.search( namespace="articles-en", top_k=5, score_by=[TextQuery(query="machine learning", fields=["content"])], include_fields=["title", "category"], filter={"category": {"$eq": "tech"}}, ) for doc in response.matches: print(doc._id, doc._score) .. seealso:: - :meth:`~pinecone.async_client.async_index.AsyncIndex.search` — record search for integrated-inference indexes. - :meth:`~pinecone.async_client.async_index.AsyncIndex.query` — nearest-neighbor search over vectors you provide. """ segment = _encode_document_namespace(namespace) body = _build_search_documents_body( score_by=score_by, top_k=top_k, include_fields=include_fields, filter=filter, ) logger.info("Searching documents in namespace %r with top_k=%d", namespace, top_k) response = await self._http.post( f"/namespaces/{segment}/documents/search", timeout=timeout, json=body, ) return DocumentsAdapter.to_search_response(response)
[docs] async def fetch( self, *, namespace: str, ids: Sequence[str] | None = None, filter: Mapping[str, Any] | None = None, include_fields: Sequence[str] | None = None, pagination_token: str | None = None, timeout: float | None = None, ) -> FetchDocumentsResponse: """Fetch documents from a namespace by ID or by metadata filter. Exactly one of ``ids`` or ``filter`` must be provided. A filtered fetch returns matching documents a page at a time — a page holds up to 10000 documents and the page size is fixed — with ``response.pagination`` carrying the token for the next page. Args: namespace (str): Namespace to fetch from (required, non-empty). ids: Document IDs to fetch (1-1000). IDs that do not exist are omitted from the result rather than raising. Mutually exclusive with ``filter``. filter: Non-empty metadata filter expression selecting the documents to fetch. Mutually exclusive with ``ids``. include_fields: Document fields to include in the response. ``None`` (default) returns all fields. pagination_token: Token from a previous filtered fetch response to retrieve the next page. Only valid together with ``filter``. timeout (float | None): Per-request timeout in seconds. Overrides the client-level default for this call only. Returns: :class:`FetchDocumentsResponse` with ``documents`` (a map of document ID to document), ``namespace``, ``usage``, and — for filtered fetches with more results — ``pagination``. Raises: :exc:`PineconeValueError`: If ``namespace`` is empty, both or neither of ``ids`` and ``filter`` are provided, ``filter`` is an empty dict, ``ids`` exceeds 1000 entries, or ``pagination_token`` is passed without ``filter``. :exc:`ApiError`: If the API returns an error response; the server's error text is surfaced intact. :exc:`PineconeConnectionError`: If a network-level connection fails (DNS, refused, transport error). :exc:`PineconeTimeoutError`: If the request exceeds the configured timeout. Examples: .. code-block:: python response = await idx.documents.fetch( namespace="articles-en", ids=["article-101", "article-102"], ) for doc_id, doc in response.documents.items(): print(doc_id, doc.title) response = await idx.documents.fetch( namespace="articles-en", filter={"category": {"$eq": "tech"}}, ) while response.pagination is not None: response = await idx.documents.fetch( namespace="articles-en", filter={"category": {"$eq": "tech"}}, pagination_token=response.pagination.next, ) """ segment = _encode_document_namespace(namespace) body = _build_fetch_documents_body( ids=ids, filter=filter, include_fields=include_fields, pagination_token=pagination_token, ) logger.info("Fetching documents from namespace %r", namespace) response = await self._http.post( f"/namespaces/{segment}/documents/fetch", timeout=timeout, json=body, ) return DocumentsAdapter.to_fetch_response(response)
[docs] async def delete( self, *, namespace: str, ids: Sequence[str] | None = None, filter: Mapping[str, Any] | None = None, delete_all: bool = False, timeout: float | None = None, ) -> DeleteDocumentsResponse: """Delete documents from a namespace by ID, filter, or delete-all flag. Exactly one of ``ids``, ``filter``, or ``delete_all`` must be provided. Deleting IDs that do not exist does not raise an error. Pinecone applies the delete asynchronously. For a filtered delete, ``response.matched_records`` is the point-in-time count of matching documents when the delete was accepted, not a guarantee of the number ultimately deleted. Args: namespace (str): Namespace to delete from (required, non-empty). ids: Document IDs to delete (1-1000). Mutually exclusive with ``filter`` and ``delete_all``. filter: Non-empty metadata filter expression selecting the documents to delete. Text-match operators (``$match_phrase``, ``$match_all``, ``$match_any``) are not supported here. Mutually exclusive with ``ids`` and ``delete_all``. Not available on an index with dedicated read capacity scaled to 0 replicas — scale up replicas first. delete_all (bool): If ``True``, delete all documents in the namespace. Mutually exclusive with ``ids`` and ``filter``. timeout (float | None): Per-request timeout in seconds. Overrides the client-level default for this call only. Returns: :class:`DeleteDocumentsResponse` — ``matched_records`` is populated only for filtered deletes (``None`` for by-ID and delete-all paths, and when the count could not be read in time). Raises: :exc:`PineconeValueError`: If ``namespace`` is empty, zero or more than one of ``ids``/``filter``/``delete_all`` is provided, ``filter`` is an empty dict, or ``ids`` exceeds 1000 entries. :exc:`ApiError`: If the API returns an error response; the server's error text is surfaced intact. :exc:`PineconeConnectionError`: If a network-level connection fails (DNS, refused, transport error). :exc:`PineconeTimeoutError`: If the request exceeds the configured timeout. Examples: .. code-block:: python await idx.documents.delete(namespace="articles-en", ids=["article-101"]) response = await idx.documents.delete( namespace="articles-en", filter={"category": {"$eq": "obsolete"}}, ) print(response.matched_records) await idx.documents.delete(namespace="articles-old", delete_all=True) """ segment = _encode_document_namespace(namespace) body = _build_delete_documents_body(ids=ids, filter=filter, delete_all=delete_all) logger.info("Deleting documents from namespace %r", namespace) response = await self._http.post( f"/namespaces/{segment}/documents/delete", timeout=timeout, json=body, ) return DocumentsAdapter.to_delete_response(response)
[docs] async def update( self, *, namespace: str, documents: Sequence[Mapping[str, Any] | UpdateDocumentRecord] | None = None, filter: Mapping[str, Any] | None = None, set_fields: Mapping[str, Any] | None = None, remove_fields: Sequence[str] | None = None, timeout: float | None = None, ) -> UpdateDocumentsResponse: """Apply partial updates to documents in a namespace. Documents are selected either per ID with ``documents``, or in bulk with ``filter`` plus ``set_fields`` and/or ``remove_fields``. The two shapes are mutually exclusive. Fields that are not mentioned are left unchanged, and an update naming a document that does not exist is accepted as a no-op rather than raising. Pinecone applies the update asynchronously — for a filtered update, ``response.matched_records`` is the point-in-time count of matching documents when the update was accepted, not a guarantee of the number ultimately patched. Args: namespace (str): Namespace to update in (required, non-empty). documents: Per-document patches (1-1000). Each element is a dict with an ``_id`` key or an :class:`~pinecone.models.documents.document.UpdateDocumentRecord`. Any key other than ``_id`` and ``_remove_fields`` sets a new value for that field; the names in ``_remove_fields`` are removed from the document. ``_id`` values must be unique within the request. Mutually exclusive with ``filter``, ``set_fields``, and ``remove_fields``. filter: Non-empty metadata filter expression selecting the documents to patch. Text-match operators (``$match_phrase``, ``$match_all``, ``$match_any``) are not supported here. Mutually exclusive with ``documents``. Not available on an index with dedicated read capacity scaled to 0 replicas — scale up replicas first. set_fields: Fields to set on every document matching ``filter``, and the values to set them to. Only valid with ``filter``. remove_fields: Names of the fields to remove from every document matching ``filter``. Only valid with ``filter``. timeout (float | None): Per-request timeout in seconds. Overrides the client-level default for this call only. Returns: :class:`UpdateDocumentsResponse` — ``matched_records`` is populated only for filtered updates (``None`` for per-ID updates, and when the count could not be read in time). Raises: :exc:`PineconeValueError`: If ``namespace`` is empty, ``documents`` is combined with any by-filter field, neither ``documents`` nor ``filter`` is given, ``set_fields`` or ``remove_fields`` is passed without ``filter``, ``filter`` is an empty dict or carries no patch, ``documents`` is empty or exceeds 1000 entries, or a patch is malformed — the message names the offending position. :exc:`ApiError`: If the API returns an error response; the server's error text is surfaced intact. A field value of ``None`` is rejected server-side — use ``_remove_fields`` (per-ID) or ``remove_fields`` (by-filter) to remove a field. :exc:`PineconeConnectionError`: If a network-level connection fails (DNS, refused, transport error). :exc:`PineconeTimeoutError`: If the request exceeds the configured timeout. Examples: .. code-block:: python await idx.documents.update( namespace="articles-en", documents=[ {"_id": "article-101", "title": "Updated title"}, {"_id": "article-102", "_remove_fields": ["content"]}, ], ) response = await idx.documents.update( namespace="articles-en", filter={"category": {"$eq": "news"}}, set_fields={"category": "archive"}, remove_fields=["content"], ) print(response.matched_records) """ segment = _encode_document_namespace(namespace) body = _build_update_documents_body( documents=documents, filter=filter, set_fields=set_fields, remove_fields=remove_fields, ) logger.info("Updating documents in namespace %r", namespace) response = await self._http.post( f"/namespaces/{segment}/documents/update", timeout=timeout, json=body, ) return DocumentsAdapter.to_update_response(response)
[docs] def list( self, *, namespace: str, prefix: str | None = None, limit: int | None = None, pagination_token: str | None = None, timeout: float | None = None, ) -> AsyncPaginator[ListedDocumentRecord]: """List the documents in a namespace, following pagination lazily. Returns an :class:`~pinecone.models.pagination.AsyncPaginator` that fetches pages on demand and stops when the server returns no ``pagination`` token. Documents come back in sorted order by ID, carrying only their ``_id``. Args: namespace (str): Namespace to list from (required, non-empty). prefix (str | None): Return only documents whose IDs begin with this prefix. At most 512 characters, ASCII only (``\\x01``-``\\x7F``). ``None`` lists every document. limit (int | None): Maximum number of documents the server returns **per page**, 1-100. This tunes the page size, not the total — the paginator follows every page. To stop early, break out of the ``async for`` loop. pagination_token (str | None): Token from a previous list response to resume from, rather than starting at the first page. timeout (float | None): Per-request timeout in seconds, applied to each page request. Overrides the client-level default. Returns: :class:`~pinecone.models.pagination.AsyncPaginator` over :class:`ListedDocumentRecord` objects. Supports ``async for``, :meth:`~pinecone.models.pagination.AsyncPaginator.to_list`, and :meth:`~pinecone.models.pagination.AsyncPaginator.pages`. Raises: :exc:`PineconeValueError`: If ``namespace`` is empty, ``prefix`` violates the rules above, or ``limit`` falls outside 1-100. Raised by this call, before the paginator is returned. :exc:`ApiError`: If the API returns an error response; the server's error text is surfaced intact. :exc:`PineconeConnectionError`: If a network-level connection fails (DNS, refused, transport error). :exc:`PineconeTimeoutError`: If a page request exceeds the configured timeout. Examples: .. code-block:: python async for doc in idx.documents.list(namespace="articles-en", prefix="article-1"): print(doc.id) paginator = idx.documents.list(namespace="articles-en", limit=20) async for page in paginator.pages(): print(len(page.items), page.pagination_token) """ segment = _encode_document_namespace(namespace) base = _build_list_documents_body(prefix=prefix, limit=limit, pagination_token=None) async def fetch_page(token: str | None) -> Page[ListedDocumentRecord]: body = dict(base) if token is not None: body["pagination_token"] = token logger.info("Listing documents in namespace %r", namespace) response = await self._http.post( f"/namespaces/{segment}/documents/list", timeout=timeout, json=body, ) result = DocumentsAdapter.to_list_response(response) next_token = result.pagination.next if result.pagination is not None else None return Page(items=result.documents, pagination_token=next_token) return AsyncPaginator(fetch_page=fetch_page, initial_token=pagination_token)