"""Indexes namespace — schema-based control-plane operations (2026-07 API)."""
from __future__ import annotations
import logging
import time
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any
from urllib.parse import quote
from pinecone._internal.adapters.backups_adapter import BackupsAdapter
from pinecone._internal.adapters.indexes_adapter import IndexesAdapter
from pinecone._internal.backups_helpers import backup_list_params
from pinecone._internal.index_migration import (
reject_integrated_spec_create,
reject_legacy_configure_kwargs,
reject_legacy_create_kwargs,
)
from pinecone._internal.indexes_helpers import poll_index_until_ready, resolve_enum_value
from pinecone._internal.legacy_index_translation import (
legacy_pod_scaling,
legacy_vector_schema,
spec_to_deployment,
spec_to_read_capacity,
)
from pinecone._internal.validation import (
require_non_empty,
require_one_of,
require_positive,
require_valid_resource_name,
validate_index_tags,
)
from pinecone.errors.exceptions import (
NotFoundError,
PineconeTimeoutError,
PineconeValueError,
)
from pinecone.models.backups.model import BackupModel
from pinecone.models.enums import Metric, PodType, VectorType
from pinecone.models.indexes.index import IndexModel
from pinecone.models.indexes.requests import ConfigureIndexRequest, CreateIndexRequest
from pinecone.models.indexes.schema import IndexSchema
from pinecone.models.pagination import Page, Paginator
if TYPE_CHECKING:
from pinecone._internal.http_client import HTTPClient
logger = logging.getLogger(__name__)
_POLL_INTERVAL_SECONDS = 5
_JSON_HEADERS = {"Content-Type": "application/json"}
_DELETION_PROTECTION_VALUES = ("disabled", "enabled")
def _require_fields_dict(param: str, schema: dict[str, Any] | IndexSchema) -> None:
"""Require a ``{"fields": {...}}``-shaped schema with at least one field."""
if isinstance(schema, IndexSchema):
return
if not schema:
raise PineconeValueError(f"{param} cannot be an empty dict")
fields = schema.get("fields")
if not isinstance(fields, dict) or "fields" not in schema:
raise PineconeValueError(
f"{param} must be a dict of the form {{'fields': {{'<field-name>': "
"{...}}}} mapping field names to searched-field configurations "
"(dense_vector, sparse_vector, or string with full_text_search). "
"Metadata-only fields are not declared at create time in the 2026-07 "
"API; they are indexed automatically at upsert."
)
if not fields:
raise PineconeValueError(
f"{param}['fields'] cannot be empty: the 2026-07 API requires at least "
"one searched field (dense_vector, sparse_vector, or string with "
"full_text_search)."
)
def _require_non_empty_dict(param: str, value: dict[str, Any] | None) -> None:
if value is not None and not value:
raise PineconeValueError(f"{param} cannot be an empty dict")
def _reject_legacy_metadata_schema(schema: dict[str, Any] | IndexSchema) -> None:
"""Raise a guided error for a 9.x metadata ``schema=`` value.
On 9.x, ``create(schema=...)`` declared filterable metadata fields shaped
``{"fields": {"<name>": {"filterable": bool}}}`` — no ``"type"`` key. At
2026-07 the same keyword declares searched fields instead, so silently
reinterpreting one as the other would create a wrong index rather than
fail loudly.
"""
if not isinstance(schema, dict):
return
fields = schema.get("fields")
if not isinstance(fields, dict) or not fields:
return
if all(isinstance(f, dict) and "type" not in f for f in fields.values()):
raise PineconeValueError(
f"schema={schema!r} looks like a 9.x metadata schema: none of its fields "
"carry a 'type' key. At 2026-07, schema= declares searched fields instead "
"(dense_vector, sparse_vector, or string with full_text_search) — metadata "
"fields are no longer declared at create time; they are indexed "
"automatically at upsert. If you meant to declare a searched field, add "
"'type': 'dense_vector' | 'sparse_vector' | 'string' to each field. See "
"the migration guide: docs/migration/v10-migration.md"
)
[docs]
class Indexes:
"""Control-plane operations for Pinecone indexes (2026-07 API).
Provides ``list``, ``describe``, ``exists``, ``create``,
``create_for_model``, ``delete``, and ``configure`` methods, plus the
index-scoped backup methods ``create_backup``, ``list_backups``, and
``describe_backup``.
.. versionchanged:: 10.0
Graduated to the 2026-07 schema-based API. ``create()`` takes
``schema=``/``deployment=`` instead of ``spec=``/``dimension=``/
``metric=``/``vector_type=``; ``configure()`` nests pod scaling under
``deployment=`` and removed ``embed=``; ``list()`` returns a
:class:`~pinecone.models.pagination.Paginator`; the index-scoped
backup methods graduated from the preview namespace.
.. seealso::
:class:`~pinecone.client.backups.Backups` (``pc.backups``) covers the
project-wide backup listing plus ``delete``, which are not scoped to
one index.
Use :meth:`Pinecone.index(name) <pinecone.Pinecone.index>` to get a
data-plane client for vector operations on a specific index.
Args:
http (HTTPClient): HTTP client for making API requests.
Examples:
.. code-block:: python
from pinecone import Pinecone
pc = Pinecone(api_key="your-api-key")
names = [idx.name for idx in pc.indexes.list()]
"""
[docs]
def __init__(self, http: HTTPClient, host_cache: dict[str, str] | None = None) -> None:
self._http = http
self._adapter = IndexesAdapter()
self._host_cache: dict[str, str] = host_cache if host_cache is not None else {}
def __repr__(self) -> str:
"""Return developer-friendly representation."""
return "Indexes()"
[docs]
def list(
self,
*,
limit: int | None = None,
pagination_token: str | None = None,
) -> Paginator[IndexModel]:
"""List all indexes in the project.
The server currently returns every index in one page, so the
returned :class:`~pinecone.models.pagination.Paginator` yields once
and stops. It still exposes the paginator interface for consistency
with other list methods, and so a future page size increase or
signature change isn't needed if the server starts paginating.
.. versionchanged:: 10.0
Returns a :class:`~pinecone.models.pagination.Paginator` instead of
an ``IndexList``. Iteration keeps working; replace
``pc.indexes.list().names()`` with
``[idx.name for idx in pc.indexes.list()]``.
Args:
limit: Maximum number of items to yield. 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:`IndexModel` instances.
Raises:
:exc:`PineconeValueError`: If *limit* is zero or negative.
:exc:`ApiError`: If the API returns an error response.
Examples:
>>> for index in pc.indexes.list(): # doctest: +SKIP
... print(index.name)
"""
if limit is not None:
require_positive("limit", limit)
def fetch_page(token: str | None) -> Page[IndexModel]:
logger.info("Listing indexes")
response = self._http.get("/indexes")
items = list(self._adapter.to_index_list(response.content))
logger.debug("Listed %d indexes", len(items))
return Page(items=items, pagination_token=None)
return Paginator(fetch_page=fetch_page, initial_token=pagination_token, limit=limit)
[docs]
def describe(self, name: str) -> IndexModel:
"""Get detailed information about a named index.
Caches the index's host internally, so a later
:meth:`Pinecone.index(name) <pinecone.Pinecone.index>` call for the
same name skips its own describe round trip.
Args:
name (str): The name of the index to describe.
Returns:
:class:`IndexModel` with name, host, schema, deployment,
read_capacity, status, deletion_protection, and tags.
Raises:
:exc:`PineconeValueError`: If *name* is empty.
:exc:`NotFoundError`: If the index does not exist.
:exc:`ApiError`: If the API returns another error response.
Examples:
>>> desc = pc.indexes.describe("my-index")
>>> desc.host # doctest: +SKIP
'https://my-index.svc.pinecone.io'
"""
require_non_empty("name", name)
logger.info("Describing index %r", name)
response = self._http.get(f"/indexes/{quote(name, safe='')}")
model = self._adapter.to_index_model(response.content)
if model.host is not None:
self._host_cache[name] = model.host
logger.debug("Described index %r (host=%s)", name, model.host)
return model
[docs]
def exists(self, name: str) -> bool:
"""Check whether a named index exists.
Calls :meth:`describe` internally and returns ``False`` instead of
raising when the index isn't found.
.. versionchanged:: 10.0
An empty *name* now raises :exc:`PineconeValueError` instead of
returning ``False``.
Args:
name (str): The name of the index to check.
Returns:
True if the index exists, False otherwise.
Raises:
:exc:`PineconeValueError`: If *name* is empty.
:exc:`ApiError`: If the API returns an error response other than a not-found error.
Examples:
>>> pc.indexes.exists("my-index") # doctest: +SKIP
True
"""
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 an index by name.
After sending the delete request, removes the cached host URL
for the index. By default, polls every 5 seconds until the index
disappears with no upper time bound.
Args:
name (str): The name of the index to delete.
timeout (int | None): Seconds to wait for the index to disappear.
Use ``None`` (default) to poll indefinitely until the index
is gone. Use a positive int to poll with a deadline.
Use ``-1`` to return immediately without polling.
Raises:
:exc:`PineconeValueError`: If *name* is empty.
:exc:`NotFoundError`: If the index does not exist.
:exc:`ForbiddenError`: If deletion protection is enabled on the index.
:exc:`PineconeTimeoutError`: If the index still exists after *timeout* seconds.
:exc:`ApiError`: If the API returns another error response.
Examples:
.. code-block:: python
pc.indexes.delete("my-index")
# Wait up to 60 seconds for deletion to complete
pc.indexes.delete("my-index", timeout=60)
"""
require_non_empty("name", name)
logger.info("Deleting index %r", name)
self._http.delete(f"/indexes/{quote(name, safe='')}")
self._host_cache.pop(name, None)
logger.debug("Deleted index %r", name)
if timeout == -1:
return
start = time.monotonic()
while True:
try:
self.describe(name)
except NotFoundError:
self._host_cache.pop(name, None)
return
if timeout is not None:
elapsed = time.monotonic() - start
if elapsed >= timeout:
raise PineconeTimeoutError(f"Index '{name}' still exists after {timeout}s")
time.sleep(_POLL_INTERVAL_SECONDS)
[docs]
def create(
self,
*,
schema: dict[str, Any] | IndexSchema | None = None,
name: str | None = None,
deployment: dict[str, Any] | None = None,
read_capacity: dict[str, Any] | None = None,
deletion_protection: str | None = None,
tags: Mapping[str, str] | None = None,
cmek_id: str | None = None,
timeout: int | None = None,
spec: Any = None,
dimension: int | None = None,
metric: Metric | str | None = None,
vector_type: VectorType | str | None = None,
**legacy_kwargs: Any,
) -> IndexModel:
"""Create a new index (2026-07 schema-based API).
An index's field layout is declared as a ``schema`` of named, typed
fields. Every field in the schema must be one that gets searched —
``dense_vector``, ``sparse_vector``, or ``string`` with
``full_text_search`` enabled. Metadata-only fields aren't declared
here; they're indexed automatically the first time they appear on an
upserted record. The schema can't change after the index is created.
.. versionchanged:: 10.0
Replaces the 2025-10 signature. ``spec=``, ``dimension=``,
``metric=``, and ``vector_type=`` are deprecated, keyword-only
sugar for the current ``schema=``/``deployment=`` arguments (see
below). ``pods=``, ``metadata_config=``, ``source_collection=``,
``source_backup_id=``, and ``spec=IntegratedSpec(...)`` have no
equivalent here; use :meth:`create_for_model` for integrated
embedding.
Args:
schema: The index's field schema. Required unless the
deprecated ``dimension=`` (with optional ``metric=``/
``vector_type=``) is used instead — the two are mutually
exclusive. A dict with a ``"fields"`` key mapping field
names to typed configurations::
{
"fields": {
"embedding": {"type": "dense_vector",
"dimension": 1536, "metric": "cosine"},
"body": {"type": "string",
"full_text_search": {"language": "en"}},
}
}
Also accepts the dict produced by
:class:`~pinecone.schema_builder.SchemaBuilder` or an
:class:`~pinecone.models.indexes.schema.IndexSchema`.
A **hybrid** index must declare its ``sparse_vector`` field
explicitly. At 2026-07 a dense field with
``metric="dotproduct"`` no longer accepts sparse values on its
own: the create succeeds, and only the sparse upserts are
refused later. The field cannot be added by ``configure()``, so
an index created without one has to be recreated. See
``docs/migration/v10-migration.md``.
``full_text_search.language`` accepts a fixed set of language
codes (or their English names, default ``en``), but
``stop_words=True`` is not supported for every language — the
server's 400 names the unsupported language, by its English
name rather than the code you sent.
name: Name for the index — 1-45 characters, lowercase
alphanumerics and hyphens (e.g. ``"movie-recommendations"``).
The server assigns a name when omitted.
deployment: Deployment configuration, discriminated on
``"deployment_type"``. For a managed index:
``{"deployment_type": "managed", "cloud": "aws", "region":
"us-east-1"}``. For a pod-based index, ``"deployment_type":
"pod"`` plus ``environment``, ``pod_type``, ``replicas``,
and ``shards``. Defaults to a managed index on AWS
``us-east-1`` when omitted. Mutually exclusive with the
deprecated ``spec=``.
read_capacity: Read capacity for a managed or BYOC index —
``{"mode": "OnDemand"}`` or ``{"mode": "Dedicated",
"dedicated": {"node_type": ..., "scaling": ..., "manual":
{"replicas": ..., "shards": ...}}}``.
deletion_protection: ``"enabled"`` to block :meth:`delete` on
this index until it's set back to ``"disabled"`` (the
default).
tags: Key-value tags to attach, e.g. ``{"env": "prod"}``, up to
20 pairs. Pass ``None`` (the default) to attach none.
cmek_id: ID of a customer-managed encryption key to encrypt the
index with.
timeout: How long to wait, in seconds, for the index to become
ready before returning. ``None`` (default) waits
indefinitely; ``-1`` returns immediately without waiting.
spec: **Deprecated.** A
:class:`~pinecone.models.indexes.specs.ServerlessSpec`,
:class:`~pinecone.models.indexes.specs.PodSpec`,
:class:`~pinecone.models.indexes.specs.ByocSpec`, or the
equivalent dict, translated into ``deployment=`` (and
``read_capacity=`` when the spec carries one). Mutually
exclusive with ``deployment=``. Use :meth:`create_for_model`
for ``IntegratedSpec``.
.. deprecated:: 10.0
Pass ``deployment=`` directly instead.
dimension: **Deprecated.** Dense vector width for the legacy
path, translated into a single-field ``schema=``. Required
when creating a dense index this way.
.. deprecated:: 10.0
Declare a named field in ``schema=`` instead.
metric: **Deprecated.** Similarity metric for the legacy dense
path — ``"cosine"`` (default), ``"euclidean"``, or
``"dotproduct"``.
.. deprecated:: 10.0
Set ``metric`` inside the ``schema=`` field declaration
instead.
vector_type: **Deprecated.** ``"dense"`` (default) or
``"sparse"``, for the legacy path.
.. deprecated:: 10.0
Declare a named ``dense_vector``/``sparse_vector`` field
in ``schema=`` instead.
Returns:
:class:`IndexModel` describing the created index — ready,
unless ``timeout=-1`` was passed.
Raises:
:exc:`PineconeValueError`: If neither ``schema=`` nor
``dimension=`` is given, or mutually exclusive arguments
(``schema=`` with a legacy vector kwarg, or ``deployment=``
with ``spec=``) are combined.
:exc:`PineconeTypeError`: If an unsupported legacy keyword (e.g.
``pods=``) or ``spec=IntegratedSpec(...)`` is passed.
:exc:`IndexInitFailedError`: If the index fails to initialize.
:exc:`PineconeTimeoutError`: If the index isn't ready before
*timeout* elapses.
:exc:`ApiError`: If the API returns another error response.
Examples:
>>> pc.indexes.create( # doctest: +SKIP
... name="movie-recommendations",
... schema={"fields": {"embedding": {
... "type": "dense_vector", "dimension": 1536, "metric": "cosine"}}},
... )
"""
reject_legacy_create_kwargs(legacy_kwargs, name)
reject_integrated_spec_create(spec, name)
legacy_vector_kwargs_given = (
dimension is not None or metric is not None or vector_type is not None
)
if schema is not None and legacy_vector_kwargs_given:
conflicting = [
kw
for kw, val in (
("dimension", dimension),
("metric", metric),
("vector_type", vector_type),
)
if val is not None
]
raise PineconeValueError(
"create() got both schema= and "
f"{', '.join(f'{kw}=' for kw in conflicting)}: schema= is the 2026-07 "
"field declaration and cannot be combined with the deprecated "
"dimension=/metric=/vector_type= sugar. Pass one or the other."
)
if deployment is not None and spec is not None:
raise PineconeValueError(
"create() got both deployment= and spec=: deployment= is the 2026-07 "
"deployment configuration and cannot be combined with the deprecated "
"spec= sugar. Pass one or the other."
)
if schema is None and not legacy_vector_kwargs_given:
raise PineconeValueError(
"schema is required: pass schema={'fields': {'<field-name>': {...}}} "
"declaring at least one searched field (dense_vector, sparse_vector, "
"or string with full_text_search), or, for backward compatibility, "
"the deprecated dimension= keyword argument (optionally with metric= "
"and vector_type=)."
)
resolved_schema: dict[str, Any] | IndexSchema
if schema is None:
resolved_schema = legacy_vector_schema(
dimension=dimension, metric=metric, vector_type=vector_type
)
else:
_reject_legacy_metadata_schema(schema)
resolved_schema = schema
_require_fields_dict("schema", resolved_schema)
if name is not None:
require_valid_resource_name("name", name)
resolved_deployment = deployment
resolved_read_capacity = read_capacity
if spec is not None:
resolved_deployment = spec_to_deployment(spec)
if resolved_read_capacity is None:
resolved_read_capacity = spec_to_read_capacity(spec)
_require_non_empty_dict("deployment", resolved_deployment)
_require_non_empty_dict("read_capacity", resolved_read_capacity)
if tags is not None and not tags:
raise PineconeValueError("tags cannot be an empty dict")
validate_index_tags(tags)
resolved_dp: str | None = None
if deletion_protection is not None:
resolved_dp = resolve_enum_value(deletion_protection)
require_one_of("deletion_protection", resolved_dp, _DELETION_PROTECTION_VALUES)
request = CreateIndexRequest(
schema=resolved_schema,
name=name,
deployment=resolved_deployment,
read_capacity=resolved_read_capacity,
deletion_protection=resolved_dp,
tags=dict(tags) if tags is not None else None,
cmek_id=cmek_id,
)
logger.info("Creating index name=%r", name)
response = self._http.post(
"/indexes",
content=self._adapter.to_create_request(request),
headers=_JSON_HEADERS,
)
model = self._adapter.to_index_model(response.content)
logger.debug("Created index %r", model.name)
if timeout != -1:
model = self._poll_until_ready(model.name, timeout)
return model
[docs]
def create_for_model(
self,
*,
name: str,
cloud: str,
region: str,
embed: Mapping[str, Any] | Any,
deletion_protection: str | None = None,
tags: Mapping[str, str] | None = None,
schema: dict[str, Any] | None = None,
read_capacity: dict[str, Any] | None = None,
timeout: int | None = None,
) -> IndexModel:
"""Create a serverless index with an integrated embedding model.
Pinecone embeds text written to the mapped field automatically at
upsert time and embeds queries at read time using the same model. In
the returned index, the embedding configuration surfaces as a
``semantic_text`` field in ``schema``, named after the ``field_map``
text entry.
Args:
name: Required name for the index (1-45 characters,
``^[a-z0-9]([a-z0-9-]*[a-z0-9])?$``).
cloud: Public cloud provider — ``"aws"``, ``"gcp"``, or ``"azure"``.
region: Cloud region (e.g. ``"us-east-1"``).
embed: Embedding configuration. A dict (or
:class:`~pinecone.models.indexes.specs.EmbedConfig` /
:class:`~pinecone.inference.models.index_embed.IndexEmbed`)
with required ``model`` and ``field_map`` (e.g.
``{"text": "chunk_text"}``) and optional ``metric``,
``dimension``, ``read_parameters``, ``write_parameters``.
The model cannot be changed after creation.
deletion_protection: ``"enabled"`` or ``"disabled"``.
tags: Optional key-value tags (same limits as :meth:`create`).
schema: Optional metadata schema dict for filterable metadata
fields, e.g. ``{"fields": {"genre": {"filterable": True}}}``.
A bare field map is wrapped in ``{"fields": ...}``.
read_capacity: Optional read capacity dict (see :meth:`create`).
timeout: Readiness polling — same semantics as :meth:`create`.
Returns:
:class:`IndexModel` describing the created index.
Raises:
:exc:`PineconeValueError`: If *name*, *cloud*, *region*, or
*embed* fail client-side validation.
:exc:`ApiError`: If the API returns an error response.
Examples:
>>> pc.indexes.create_for_model( # doctest: +SKIP
... name="semantic-search",
... cloud="aws",
... region="us-east-1",
... embed={"model": "multilingual-e5-large",
... "field_map": {"text": "chunk_text"}},
... )
"""
require_valid_resource_name("name", name)
cloud_str = resolve_enum_value(cloud)
region_str = resolve_enum_value(region)
require_non_empty("cloud", cloud_str)
require_non_empty("region", region_str)
embed_body = self._embed_to_body(embed)
if tags is not None and not tags:
raise PineconeValueError("tags cannot be an empty dict")
validate_index_tags(tags)
_require_non_empty_dict("read_capacity", read_capacity)
body: dict[str, Any] = {
"name": name,
"cloud": cloud_str,
"region": region_str,
"embed": embed_body,
}
if deletion_protection is not None:
resolved_dp = resolve_enum_value(deletion_protection)
require_one_of("deletion_protection", resolved_dp, _DELETION_PROTECTION_VALUES)
body["deletion_protection"] = resolved_dp
if tags is not None:
body["tags"] = dict(tags)
if schema is not None:
if not schema:
raise PineconeValueError("schema cannot be an empty dict")
wrapped = schema if list(schema.keys()) == ["fields"] else {"fields": schema}
body["schema"] = wrapped
if read_capacity is not None:
body["read_capacity"] = read_capacity
logger.info("Creating integrated index name=%r", name)
response = self._http.post("/indexes/create-for-model", json=body)
model = self._adapter.to_index_model(response.content)
logger.debug("Created integrated index %r", model.name)
if timeout != -1:
model = self._poll_until_ready(model.name, timeout)
return model
[docs]
def create_backup(
self,
index_name: str,
*,
name: str | None = None,
description: str | None = None,
) -> BackupModel:
"""Create a backup of an index.
Index-scoped shortcut for :meth:`Pinecone.backups.create` — pass the
same arguments either way.
.. versionadded:: 10.0
Graduated from ``pc.preview.indexes.create_backup``, now returning
the single top-level
:class:`~pinecone.models.backups.model.BackupModel`.
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.models.backups.model.BackupModel` describing the
new backup. ``status`` is typically ``"Initializing"`` right after
creation; poll :meth:`describe_backup` until it reads ``"Ready"``
before restoring from it.
Raises:
:exc:`PineconeValueError`: If *index_name* is empty.
:exc:`NotFoundError`: If the index does not exist.
:exc:`ApiError`: If the API returns another error response.
Examples:
>>> backup = pc.indexes.create_backup("my-index", name="nightly") # doctest: +SKIP
>>> backup.status # doctest: +SKIP
'Initializing'
"""
require_non_empty("index_name", index_name)
body: dict[str, str] = {}
if name is not None:
body["name"] = name
if description is not None:
body["description"] = description
logger.info("Creating backup for index %r", index_name)
response = self._http.post(f"/indexes/{quote(index_name, safe='')}/backups", json=body)
result = BackupsAdapter.to_backup(response.content)
logger.debug("Created backup %r", result.backup_id)
return result
[docs]
def list_backups(
self,
index_name: str,
*,
limit: int | None = None,
pagination_token: str | None = None,
include_deleted: bool | None = None,
) -> Paginator[BackupModel]:
"""List the backups of one index.
.. versionadded:: 10.0
Graduated from ``pc.preview.indexes.list_backups``, and gained
*include_deleted*. For the project-wide listing use
:meth:`Pinecone.backups.list` with no ``index_name``.
.. important::
:exc:`NotFoundError` here does not necessarily mean *index_name*
was never used. With *include_deleted* omitted or ``False``,
*index_name* must resolve to an **active** index: if every index
that used the name has since been deleted, this raises
:exc:`NotFoundError` rather than returning an empty list. Retry
with ``include_deleted=True`` to get those backups back; a
:exc:`NotFoundError` there means the name was never used in this
project.
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. It also
sets the requested page size, but only on a request that
carries no pagination token: every later page is sized by the
token, which already encodes it.
pagination_token: Token to resume pagination from a previous call.
*limit* still caps the total yield, but it is not sent
alongside a token — see above.
include_deleted: When ``True``, include backups of every index
that has ever used *index_name*, deleted ones included; those
backups carry a non-``None``
:attr:`~pinecone.models.backups.model.BackupModel.source_index_deleted_at`.
When ``None`` (the default) the parameter is omitted entirely
and the server's default (``false``) applies.
Returns:
:class:`~pinecone.models.pagination.Paginator` over
:class:`~pinecone.models.backups.model.BackupModel` instances.
Iteration stops when the response carries no pagination envelope.
Raises:
:exc:`PineconeValueError`: If *index_name* is empty or *limit* is
zero or negative.
:exc:`NotFoundError`: If *index_name* does not resolve to an
active index — see above.
:exc:`ApiError`: If the API returns another error response.
Examples:
>>> for backup in pc.indexes.list_backups("my-index"): # doctest: +SKIP
... print(backup.backup_id, backup.status)
>>> orphans = pc.indexes.list_backups( # doctest: +SKIP
... "my-index", include_deleted=True
... )
>>> [b.backup_id for b in orphans if b.source_index_deleted_at] # doctest: +SKIP
['bkp_oldidx']
"""
require_non_empty("index_name", index_name)
if limit is not None:
require_positive("limit", limit)
def fetch_page(token: str | None) -> Page[BackupModel]:
params = backup_list_params(
limit=limit,
pagination_token=token,
include_deleted=include_deleted,
)
logger.info("Listing backups for index %r", index_name)
response = self._http.get(
f"/indexes/{quote(index_name, safe='')}/backups", params=params
)
result = BackupsAdapter.to_backup_list(response.content)
next_token = result.pagination.next if result.pagination is not None else None
return Page(items=list(result), pagination_token=next_token)
return Paginator(fetch_page=fetch_page, initial_token=pagination_token, limit=limit)
[docs]
def describe_backup(self, backup_id: str) -> BackupModel:
"""Describe a backup by its ID.
Alias of :meth:`Pinecone.backups.describe`. Backups are identified
independently of any index, so despite living on ``indexes`` this
takes a backup ID rather than an index name.
.. versionadded:: 10.0
Graduated from ``pc.preview.indexes.describe_backup``.
Args:
backup_id: The unique identifier of the backup to describe.
Returns:
:class:`~pinecone.models.backups.model.BackupModel` with the
current state of the backup.
Raises:
:exc:`PineconeValueError`: If *backup_id* is empty.
:exc:`NotFoundError`: If the backup does not exist.
:exc:`ApiError`: If the API returns another error response.
Examples:
>>> backup = pc.indexes.describe_backup("bkp-123") # doctest: +SKIP
>>> backup.status # doctest: +SKIP
'Ready'
"""
require_non_empty("backup_id", backup_id)
logger.info("Describing backup %r", backup_id)
response = self._http.get(f"/backups/{quote(backup_id, safe='')}")
return BackupsAdapter.to_backup(response.content)
@staticmethod
def _embed_to_body(embed: Mapping[str, Any] | Any) -> dict[str, Any]:
"""Normalize dict / EmbedConfig / IndexEmbed into the embed wire dict."""
if isinstance(embed, Mapping):
raw: dict[str, Any] = dict(embed)
else:
raw = {
"model": getattr(embed, "model", None),
"field_map": getattr(embed, "field_map", None),
"metric": getattr(embed, "metric", None),
"dimension": getattr(embed, "dimension", None),
"read_parameters": getattr(embed, "read_parameters", None),
"write_parameters": getattr(embed, "write_parameters", None),
}
model = raw.get("model")
field_map = raw.get("field_map")
if not model:
raise PineconeValueError(
"embed must include a non-empty 'model' (e.g. 'multilingual-e5-large')"
)
if not field_map:
raise PineconeValueError(
"embed must include a non-empty 'field_map' "
"(e.g. {'text': 'chunk_text'}) naming the text field to embed"
)
body: dict[str, Any] = {
"model": resolve_enum_value(model),
"field_map": dict(field_map),
}
for key in ("metric", "dimension", "read_parameters", "write_parameters"):
value = raw.get(key)
if value:
body[key] = resolve_enum_value(value) if key == "metric" else value
return body
def _poll_until_ready(self, name: str, timeout: int | None) -> IndexModel:
"""Poll describe() until the index is ready or timeout is reached."""
return poll_index_until_ready(self.describe, name, timeout)