Source code for pinecone.preview.indexes

"""Preview indexes namespace — control-plane operations backed by 2026-01.alpha."""

from __future__ import annotations

import logging
import time
from typing import TYPE_CHECKING, Any

import msgspec

from pinecone._internal.constants import DEFAULT_BASE_URL
from pinecone._internal.validation import require_non_empty, require_positive
from pinecone.errors.exceptions import NotFoundError, PineconeTimeoutError, PineconeValueError
from pinecone.models.pagination import Page, Paginator
from pinecone.preview._internal.adapters.backups import (
    PreviewDescribeBackupAdapter,
    PreviewListBackupsAdapter,
)
from pinecone.preview._internal.adapters.indexes import (
    PreviewConfigureIndexAdapter,
    PreviewCreateIndexAdapter,
    PreviewDescribeIndexAdapter,
    PreviewListIndexesAdapter,
)
from pinecone.preview._internal.constants import INDEXES_API_VERSION
from pinecone.preview.models.backups import PreviewBackupModel, PreviewCreateBackupRequest
from pinecone.preview.models.indexes import PreviewIndexModel
from pinecone.preview.models.requests import PreviewConfigureIndexRequest, PreviewCreateIndexRequest

if TYPE_CHECKING:
    from pinecone._internal.config import PineconeConfig

logger = logging.getLogger(__name__)

_POLL_INTERVAL_SECONDS = 5

__all__ = ["PreviewIndexes"]


[docs] class PreviewIndexes: """Control-plane operations for preview indexes (``2026-01.alpha``). .. admonition:: Preview :class: warning Uses Pinecone API version ``2026-01.alpha``. Preview surface is not covered by SemVer — signatures and behavior may change in any minor SDK release. Pin your SDK version when relying on preview features. Access via ``pc.preview.indexes``. Args: config: SDK configuration used to construct an HTTP client targeting the preview API version. """
[docs] def __init__(self, config: PineconeConfig) -> None: from pinecone._internal.config import PineconeConfig as _PineconeConfig from pinecone._internal.http_client import HTTPClient as _HTTPClient self._config = config cp_host = (config.host or DEFAULT_BASE_URL).rstrip("/") cp_config = _PineconeConfig( api_key=config.api_key, host=cp_host, timeout=config.timeout, additional_headers=config.additional_headers, source_tag=config.source_tag or "", proxy_url=config.proxy_url or "", proxy_headers=config.proxy_headers, ssl_ca_certs=config.ssl_ca_certs, ssl_verify=config.ssl_verify, connection_pool_maxsize=config.connection_pool_maxsize, retry_config=config.retry_config, ) self._http = _HTTPClient(cp_config, INDEXES_API_VERSION)
[docs] def close(self) -> None: """Close the underlying HTTP client. Idempotent. .. admonition:: Preview :class: warning Uses Pinecone API version ``2026-01.alpha``. Preview surface is not covered by SemVer — signatures and behavior may change in any minor SDK release. Pin your SDK version when relying on preview features. """ self._http.close()
def __repr__(self) -> str: return "PreviewIndexes()"
[docs] def create( self, *, schema: dict[str, Any], name: str | None = None, deployment: dict[str, Any] | None = None, read_capacity: dict[str, Any] | None = None, deletion_protection: str | None = None, tags: dict[str, str] | None = None, ) -> PreviewIndexModel: """Create a new preview index. .. admonition:: Preview :class: warning Uses Pinecone API version ``2026-01.alpha``. Preview surface is not covered by SemVer — signatures and behavior may change in any minor SDK release. Pin your SDK version when relying on preview features. Args: schema: Index schema definition. A dict with a ``"fields"`` key mapping field names to their typed field definitions. Example:: { "fields": { "embedding": {"type": "dense_vector", "dimension": 1536}, "title": {"type": "string", "full_text_search": {}}, } } name: Optional name for the index. If omitted, the API assigns one. deployment: Optional deployment configuration dict. Must include a ``"deployment_type"`` key (e.g. ``"managed"``, ``"pod"``, ``"byoc"``). When omitted the API uses its default deployment. read_capacity: Optional read capacity configuration dict. Must include a ``"mode"`` key (``"OnDemand"`` or ``"Dedicated"``). When omitted the API uses on-demand mode. deletion_protection: Optional deletion-protection setting — ``"enabled"`` or ``"disabled"``. Defaults to ``"disabled"`` if not provided. tags: Optional key-value tags for the index. Keys must be at most 80 characters; values must be at most 120 characters. Returns: :class:`PreviewIndexModel` describing the newly created index. The returned model's ``status.ready`` may be ``False`` — use ``describe()`` to poll until the index is ready. Raises: :exc:`~pinecone.errors.exceptions.PineconeValueError`: If a tag key exceeds 80 characters or a tag value exceeds 120 characters. :exc:`~pinecone.errors.exceptions.ApiError`: If the API returns an error response. Examples: .. code-block:: python pc.preview.indexes.create( schema={"fields": {"embedding": {"type": "dense_vector", "dimension": 1536}}}, name="my-preview-index", ) """ if tags is not None: for key, value in tags.items(): if len(key) > 80: raise PineconeValueError(f"Tag key {key!r} exceeds the 80-character limit.") if len(value) > 120: raise PineconeValueError( f"Tag value for key {key!r} exceeds the 120-character limit." ) req = PreviewCreateIndexRequest( schema=schema, name=name, deployment=deployment, read_capacity=read_capacity, deletion_protection=deletion_protection, tags=tags, ) logger.info("Creating preview index name=%r", name) response = self._http.post( "/indexes", content=PreviewCreateIndexAdapter.to_request(req), headers={"Content-Type": "application/json"}, ) return PreviewCreateIndexAdapter.from_response(response.content)
[docs] def configure( self, name: str, *, schema: dict[str, Any] | None = None, deletion_protection: str | None = None, tags: dict[str, str] | None = None, read_capacity: dict[str, Any] | None = None, ) -> PreviewIndexModel: """Update configuration of an existing preview index. .. admonition:: Preview :class: warning Uses Pinecone API version ``2026-01.alpha``. Preview surface is not covered by SemVer — signatures and behavior may change in any minor SDK release. Pin your SDK version when relying on preview features. Only the fields you provide are updated; omitted parameters are left unchanged on the server. Args: name: Name of the preview index to configure. schema: Updated schema definition. Pass a dict with a ``"fields"`` key mapping field names to typed field definitions. The server only allows additive changes — it rejects requests that remove or modify existing fields. Example:: pc.preview.indexes.configure( "my-index", schema={"fields": { "summary": {"type": "string", "full_text_search": {}}, }}, ) deletion_protection: ``"enabled"`` to prevent accidental deletion; ``"disabled"`` to allow it. tags: Replacement set of key-value tags for the index. Keys must be at most 80 characters; values at most 120 characters. read_capacity: Updated read capacity configuration dict. Must include a ``"mode"`` key (``"OnDemand"`` or ``"Dedicated"``). Returns: :class:`PreviewIndexModel` reflecting the updated index state. Raises: :exc:`~pinecone.errors.exceptions.PineconeValueError`: If *name* is empty; if all kwargs are ``None``; if *schema*, *tags*, or *read_capacity* is an empty dict; or if a tag key/value exceeds the length limit. :exc:`~pinecone.errors.exceptions.ApiError`: If the API returns an error response. Examples: .. code-block:: python pc.preview.indexes.configure( "my-index", schema={"fields": {"summary": {"type": "string"}}}, ) Update read capacity to dedicated:: pc.preview.indexes.configure( "my-index", read_capacity={"mode": "Dedicated", "node_type": "b1", "scaling": "Manual", "manual": {"shards": 2, "replicas": 1}}, ) Update tags:: pc.preview.indexes.configure( "my-index", tags={"env": "prod", "team": "search"}, ) Enable deletion protection:: pc.preview.indexes.configure("my-index", deletion_protection="enabled") """ require_non_empty("name", name) if schema is not None and not schema: raise PineconeValueError("schema cannot be an empty dict") if tags is not None and not tags: raise PineconeValueError("tags cannot be an empty dict") if read_capacity is not None and not read_capacity: raise PineconeValueError("read_capacity cannot be an empty dict") if ( schema is None and deletion_protection is None and tags is None and read_capacity is None ): raise PineconeValueError( "at least one configuration parameter must be provided: " "schema, deletion_protection, tags, or read_capacity" ) if tags is not None: for key, value in tags.items(): if len(key) > 80: raise PineconeValueError(f"Tag key {key!r} exceeds the 80-character limit.") if len(value) > 120: raise PineconeValueError( f"Tag value for key {key!r} exceeds the 120-character limit." ) req = PreviewConfigureIndexRequest( schema=schema, read_capacity=read_capacity, deletion_protection=deletion_protection, tags=tags, ) provided = [ k for k, v in { "schema": schema, "deletion_protection": deletion_protection, "tags": tags, "read_capacity": read_capacity, }.items() if v is not None ] logger.info("Configuring preview index name=%r params=%r", name, provided) response = self._http.patch( f"/indexes/{name}", content=PreviewConfigureIndexAdapter.to_request(req), headers={"Content-Type": "application/json"}, ) return PreviewConfigureIndexAdapter.from_response(response.content)
[docs] def describe(self, name: str) -> PreviewIndexModel: """Get detailed information about a named preview index. .. admonition:: Preview :class: warning Uses Pinecone API version ``2026-01.alpha``. Preview surface is not covered by SemVer — signatures and behavior may change in any minor SDK release. Pin your SDK version when relying on preview features. Args: name: Name of the preview index to describe. Returns: :class:`PreviewIndexModel` with name, host, schema, deployment, read_capacity, status, deletion_protection, and tags. Raises: :exc:`~pinecone.errors.exceptions.PineconeValueError`: If *name* is empty. :exc:`~pinecone.errors.exceptions.NotFoundError`: If the index does not exist. :exc:`~pinecone.errors.exceptions.ApiError`: If the API returns another error response. Examples: .. code-block:: python desc = pc.preview.indexes.describe("my-preview-index") print(desc.host) """ require_non_empty("name", name) logger.info("Describing preview index name=%r", name) response = self._http.get(f"/indexes/{name}") return PreviewDescribeIndexAdapter.from_response(response.content)
[docs] def list( self, *, limit: int | None = None, pagination_token: str | None = None, ) -> Paginator[PreviewIndexModel]: """List all preview indexes. .. admonition:: Preview :class: warning Uses Pinecone API version ``2026-01.alpha``. Preview surface is not covered by SemVer — signatures and behavior may change in any minor SDK release. Pin your SDK version when relying on preview features. The 2026-01.alpha server returns all indexes in a single page. The returned :class:`~pinecone.models.pagination.Paginator` always yields exactly one page and then terminates, but the paginator interface is used for consistency with other list methods and forward compatibility. Args: limit: Maximum number of items to yield across all pages. Must be a positive integer. ``None`` yields all items. pagination_token: Token to resume pagination from a previous call. ``None`` starts from the beginning. Returns: :class:`~pinecone.models.pagination.Paginator` over :class:`~pinecone.preview.models.indexes.PreviewIndexModel` instances. Raises: :exc:`~pinecone.errors.exceptions.PineconeValueError`: If *limit* is zero or negative. :exc:`~pinecone.errors.exceptions.ApiError`: If the API returns an error response. Examples: .. code-block:: python for index in pc.preview.indexes.list(): print(index.name) Access page-level metadata:: for page in pc.preview.indexes.list().pages(): print(f"Got {len(page.items)} indexes") for index in page.items: print(index.name) """ if limit is not None: require_positive("limit", limit) def fetch_page(token: str | None) -> Page[PreviewIndexModel]: response = self._http.get("/indexes") items = PreviewListIndexesAdapter.from_response(response.content) return Page(items=items, pagination_token=None) return Paginator(fetch_page=fetch_page, initial_token=pagination_token, limit=limit)
[docs] def exists(self, name: str) -> bool: """Check whether a named preview index exists. .. admonition:: Preview :class: warning Uses Pinecone API version ``2026-01.alpha``. Preview surface is not covered by SemVer — signatures and behavior may change in any minor SDK release. Pin your SDK version when relying on preview features. Uses :meth:`describe` internally; catches :exc:`NotFoundError` and returns ``False``. Args: name: Name of the preview index to check. Returns: ``True`` if the index exists, ``False`` if it does not. Raises: :exc:`~pinecone.errors.exceptions.PineconeValueError`: If *name* is empty. :exc:`~pinecone.errors.exceptions.ApiError`: If the API returns an error other than 404. Examples: .. code-block:: python if pc.preview.indexes.exists("my-preview-index"): print("Index found") """ require_non_empty("name", name) try: self.describe(name) return True except NotFoundError: return False
[docs] def delete(self, name: str, *, timeout: int | None = None) -> None: """Delete a named preview index. .. admonition:: Preview :class: warning Uses Pinecone API version ``2026-01.alpha``. Preview surface is not covered by SemVer — signatures and behavior may change in any minor SDK release. Pin your SDK version when relying on preview features. Sends a DELETE request and, depending on *timeout*, polls until the index disappears. Args: name: Name of the preview index to delete. timeout: Controls post-delete polling behaviour. - ``None`` (default): poll :meth:`describe` every 5 seconds with no upper bound until the index is gone. - ``-1``: return immediately after the DELETE response without polling. - Positive integer: poll until the index is gone or *timeout* seconds have elapsed. Raises :exc:`PineconeTimeoutError` if the deadline is reached before the index disappears. Returns: ``None`` Raises: :exc:`~pinecone.errors.exceptions.PineconeValueError`: If *name* is empty. :exc:`~pinecone.errors.exceptions.NotFoundError`: If the index does not exist. :exc:`~pinecone.errors.exceptions.ForbiddenError`: If deletion protection is enabled on the index. :exc:`~pinecone.errors.exceptions.PineconeTimeoutError`: If *timeout* seconds elapse before the index disappears. :exc:`~pinecone.errors.exceptions.ApiError`: If the API returns another error response. Examples: .. code-block:: python pc.preview.indexes.delete("my-preview-index") # Delete and wait up to 60 seconds pc.preview.indexes.delete("my-preview-index", timeout=60) # Delete without polling pc.preview.indexes.delete("my-preview-index", timeout=-1) """ require_non_empty("name", name) logger.info("Deleting preview index name=%r", name) self._http.delete(f"/indexes/{name}") if timeout == -1: return start = time.monotonic() while True: try: self.describe(name) except NotFoundError: return if timeout is not None: elapsed = time.monotonic() - start if elapsed >= timeout: raise PineconeTimeoutError(f"Index {name!r} still exists after {timeout}s") time.sleep(_POLL_INTERVAL_SECONDS)
[docs] def create_backup( self, index_name: str, *, name: str | None = None, description: str | None = None, ) -> PreviewBackupModel: """Create a backup of a preview index. .. admonition:: Preview :class: warning Uses Pinecone API version ``2026-01.alpha``. Preview surface is not covered by SemVer — signatures and behavior may change in any minor SDK release. Pin your SDK version when relying on preview features. Args: index_name: Name of the index to back up. name: Optional user-defined name for the backup. description: Optional description providing context for the backup. Returns: :class:`~pinecone.preview.models.backups.PreviewBackupModel` describing the newly created backup. The ``status`` field will typically be ``"Initializing"`` immediately after creation. Poll ``describe_backup()`` until ``status == "Ready"`` before using the backup:: backup = pc.preview.indexes.create_backup("my-index", name="nightly") # Poll until ready import time while backup.status != "Ready": time.sleep(5) backup = pc.preview.indexes.describe_backup(backup.backup_id) Raises: :exc:`~pinecone.errors.exceptions.PineconeValueError`: If *index_name* is empty. :exc:`~pinecone.errors.exceptions.ApiError`: If the API returns an error response. Examples: .. code-block:: python pc.preview.indexes.create_backup("my-index") pc.preview.indexes.create_backup( "my-index", name="nightly", description="Daily backup" ) """ require_non_empty("index_name", index_name) if name is not None or description is not None: req = PreviewCreateBackupRequest(name=name, description=description) content = msgspec.json.encode(req) else: content = b"{}" logger.info("Creating backup for preview index index_name=%r", index_name) response = self._http.post( f"/indexes/{index_name}/backups", content=content, headers={"Content-Type": "application/json"}, ) return PreviewDescribeBackupAdapter.from_response(response.content)
[docs] def list_backups( self, index_name: str, *, limit: int | None = None, pagination_token: str | None = None, ) -> Paginator[PreviewBackupModel]: """List backups for a preview index. .. admonition:: Preview :class: warning Uses Pinecone API version ``2026-01.alpha``. Preview surface is not covered by SemVer — signatures and behavior may change in any minor SDK release. Pin your SDK version when relying on preview features. Args: index_name: Name of the index whose backups to list. limit: Maximum number of backups to yield across all pages. Must be a positive integer. ``None`` yields all backups. pagination_token: Token to resume pagination from a previous call. ``None`` starts from the beginning. Returns: :class:`~pinecone.models.pagination.Paginator` over :class:`~pinecone.preview.models.backups.PreviewBackupModel` instances. Raises: :exc:`~pinecone.errors.exceptions.PineconeValueError`: If *index_name* is empty or *limit* is zero or negative. :exc:`~pinecone.errors.exceptions.ApiError`: If the API returns an error response. Examples: .. code-block:: python for backup in pc.preview.indexes.list_backups("my-index"): print(backup.backup_id, backup.status) Access page-level metadata:: for page in pc.preview.indexes.list_backups("my-index").pages(): print(f"Got {len(page.items)} backups") for backup in page.items: print(backup.backup_id) """ require_non_empty("index_name", index_name) if limit is not None: require_positive("limit", limit) def fetch_page(token: str | None) -> Page[PreviewBackupModel]: params: dict[str, str] = {} if token: params["paginationToken"] = token response = self._http.get(f"/indexes/{index_name}/backups", params=params) items, next_token = PreviewListBackupsAdapter.from_response(response.content) return Page(items=items, pagination_token=next_token) return Paginator(fetch_page=fetch_page, initial_token=pagination_token, limit=limit)