"""Async preview indexes namespace — control-plane operations backed by 2026-01.alpha."""
from __future__ import annotations
import asyncio
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 AsyncPaginator, Page
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__ = ["AsyncPreviewIndexes"]
[docs]
class AsyncPreviewIndexes:
"""Async 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 async HTTP client targeting
the preview API version.
Examples:
.. code-block:: python
pc = AsyncPinecone(api_key="...")
index = await pc.preview.indexes.create(
schema={"fields": {"embedding": {"type": "dense_vector", "dimension": 1536}}},
name="my-preview-index",
)
"""
[docs]
def __init__(self, config: PineconeConfig) -> None:
from pinecone._internal.config import PineconeConfig as _PineconeConfig
from pinecone._internal.http_client import AsyncHTTPClient as _AsyncHTTPClient
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 = _AsyncHTTPClient(cp_config, INDEXES_API_VERSION)
[docs]
async 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.
"""
await self._http.close()
def __repr__(self) -> str:
return "AsyncPreviewIndexes()"
[docs]
async 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
index = await 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 = await self._http.post(
"/indexes",
content=PreviewCreateIndexAdapter.to_request(req),
headers={"Content-Type": "application/json"},
)
return PreviewCreateIndexAdapter.from_response(response.content)
[docs]
async 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 = await pc.preview.indexes.describe("product-search-preview")
print(desc.host)
"""
require_non_empty("name", name)
logger.info("Describing preview index name=%r", name)
response = await 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,
) -> AsyncPaginator[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.AsyncPaginator` always
yields exactly one page and then terminates, but the paginator interface
is used for consistency 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.AsyncPaginator` 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
async for index in pc.preview.indexes.list():
print(index.name)
Collect all into a list::
all_indexes = await pc.preview.indexes.list().to_list()
"""
if limit is not None:
require_positive("limit", limit)
async def fetch_page(token: str | None) -> Page[PreviewIndexModel]:
response = await self._http.get("/indexes")
items = PreviewListIndexesAdapter.from_response(response.content)
return Page(items=items, pagination_token=None)
return AsyncPaginator(fetch_page=fetch_page, initial_token=pagination_token, limit=limit)
[docs]
async 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 await pc.preview.indexes.exists("product-search-preview"):
print("Index found")
"""
require_non_empty("name", name)
try:
await self.describe(name)
return True
except NotFoundError:
return False
[docs]
async 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.
: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
await pc.preview.indexes.delete("product-search-preview")
# Delete and wait up to 60 seconds
await pc.preview.indexes.delete("product-search-preview", timeout=60)
# Delete without polling
await pc.preview.indexes.delete("product-search-preview", timeout=-1)
"""
require_non_empty("name", name)
logger.info("Deleting preview index name=%r", name)
await self._http.delete(f"/indexes/{name}")
if timeout == -1:
return
start = time.monotonic()
while True:
try:
await 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")
await asyncio.sleep(_POLL_INTERVAL_SECONDS)
[docs]
async 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 = await pc.preview.indexes.create_backup("my-index", name="nightly")
# Poll until ready
import asyncio
while backup.status != "Ready":
await asyncio.sleep(5)
backup = await 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
await pc.preview.indexes.create_backup("my-index")
await 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 = await 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,
) -> AsyncPaginator[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.AsyncPaginator` 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
async for backup in pc.preview.indexes.list_backups("my-index"):
print(backup.backup_id, backup.status)
Access page-level metadata::
paginator = pc.preview.indexes.list_backups("my-index")
async for page in paginator.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)
async def fetch_page(token: str | None) -> Page[PreviewBackupModel]:
params: dict[str, str] = {"paginationToken": token} if token is not None else {}
response = await 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 AsyncPaginator(fetch_page=fetch_page, initial_token=pagination_token, limit=limit)