"""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)