Source code for pinecone.models.indexes.requests
"""Index request models (2026-07 API).
These models are the typed boundary for index create and configure calls.
User code passes keyword arguments; the SDK validates against these models
and serialises with msgspec/orjson. Both use ``omit_defaults=True`` so
unset optional fields stay off the wire and PATCH bodies remain sparse.
"""
from __future__ import annotations
from typing import Any
from msgspec import Struct
from pinecone.errors.exceptions import PineconeValueError
from pinecone.models.indexes.deployment import IndexDeployment
from pinecone.models.indexes.schema import IndexSchema
__all__ = ["ConfigureIndexRequest", "CreateIndexRequest"]
_ALLOWED_DEPLOYMENT_TYPES = ("managed", "pod", "byoc")
_MAX_FIELD_NAME_LENGTH = 64
def validate_schema_field_name(name: str) -> None:
"""Validate a schema field name against the 2026-07 naming rules.
Field names must be 1-64 characters. Which names are reserved (e.g.
``_id``, ``_values``, ``_sparse_values``) or otherwise special (e.g. a
leading ``$``) is the server's call, not the SDK's: the 2026-07 API
accepts such names for backward compatibility, and a client-side copy
of that list would drift the moment the server's rules change.
Raises:
PineconeValueError: If the name is empty or exceeds the maximum
length. The message names the field, the rule violated, and the
fix.
"""
if not name:
raise PineconeValueError(
"Invalid schema field name '': field names must be 1-64 characters. "
"Provide a non-empty field name."
)
if len(name) > _MAX_FIELD_NAME_LENGTH:
raise PineconeValueError(
f"Invalid schema field name {name!r}: {len(name)} characters exceeds "
f"the maximum length of {_MAX_FIELD_NAME_LENGTH}. Shorten the field name."
)
def _validate_schema(schema: dict[str, Any] | IndexSchema) -> None:
fields = schema.get("fields") if isinstance(schema, dict) else schema.fields
if not isinstance(fields, dict):
return
for field_name in fields:
if not isinstance(field_name, str):
raise PineconeValueError(
f"Invalid schema field name {field_name!r}: expected a str key, "
f"got {type(field_name).__name__}. Schema field names must be strings."
)
validate_schema_field_name(field_name)
def _validate_deployment(deployment: dict[str, Any] | IndexDeployment | None) -> None:
if not isinstance(deployment, dict):
return
deployment_type = deployment.get("deployment_type")
if deployment_type is not None and deployment_type not in _ALLOWED_DEPLOYMENT_TYPES:
allowed = " | ".join(_ALLOWED_DEPLOYMENT_TYPES)
raise PineconeValueError(
f"Invalid deployment_type {deployment_type!r}: expected one of {allowed}. "
"Set deployment={'deployment_type': 'managed', 'cloud': ..., 'region': ...} "
"for a serverless index."
)
[docs]
class CreateIndexRequest(Struct, kw_only=True, omit_defaults=True):
"""Request model for creating an index.
Attributes:
schema: Index schema definition (required). Maps field names to
searched-field configurations — ``dense_vector``,
``sparse_vector``, or ``string`` with ``full_text_search``.
name: Optional name for the index. Auto-generated by the server
if omitted.
deployment: Optional deployment configuration, discriminated on
``deployment_type`` (``managed`` | ``pod`` | ``byoc``).
Defaults server-side to managed on AWS ``us-east-1``.
read_capacity: Optional read capacity configuration.
deletion_protection: Optional deletion protection setting
(``"enabled"`` or ``"disabled"``).
tags: Optional key-value tags for the index.
source_collection: Optional name of an existing collection to
create the index from.
source_backup_id: Optional ID of an existing backup to create the
index from.
cmek_id: Optional customer-managed encryption key ID (valid for
managed/BYOC indexes without full-text search fields).
Raises:
PineconeValueError: If ``deployment`` names a ``deployment_type``
that is not one of the discriminator values. The comparison is
case-sensitive, so ``"MANAGED"`` is rejected.
"""
schema: dict[str, Any] | IndexSchema
name: str | None = None
deployment: dict[str, Any] | IndexDeployment | None = None
read_capacity: dict[str, Any] | None = None
deletion_protection: str | None = None
tags: dict[str, str] | None = None
source_collection: str | None = None
source_backup_id: str | None = None
cmek_id: str | None = None
def __post_init__(self) -> None:
_validate_schema(self.schema)
_validate_deployment(self.deployment)