Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions pinecone/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,11 @@
"SemanticTextField": ("pinecone.db_control.models", "SemanticTextField"),
"SchemaField": ("pinecone.db_control.models", "SchemaField"),
"SchemaBuilder": ("pinecone.db_control.models", "SchemaBuilder"),
# Deployment classes
"ServerlessDeployment": ("pinecone.db_control.models", "ServerlessDeployment"),
"ByocDeployment": ("pinecone.db_control.models", "ByocDeployment"),
"PodDeployment": ("pinecone.db_control.models", "PodDeployment"),
"Deployment": ("pinecone.db_control.models", "Deployment"),
# Read capacity TypedDict classes
"ScalingConfigManualDict": (
"pinecone.db_control.models.serverless_spec",
Expand Down
5 changes: 5 additions & 0 deletions pinecone/db_control/models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
SchemaField,
)
from .schema_builder import SchemaBuilder
from .deployment import ServerlessDeployment, ByocDeployment, PodDeployment, Deployment


__all__ = [
Expand All @@ -46,4 +47,8 @@
"SemanticTextField",
"SchemaField",
"SchemaBuilder",
"ServerlessDeployment",
"ByocDeployment",
"PodDeployment",
"Deployment",
]
134 changes: 134 additions & 0 deletions pinecone/db_control/models/deployment.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
"""Deployment model classes for index deployment configurations.

These classes represent different deployment configurations for Pinecone indexes,
including serverless, BYOC (bring-your-own-cloud), and pod-based deployments.
"""

from __future__ import annotations

from dataclasses import dataclass


@dataclass
class ServerlessDeployment:
"""Serverless deployment configuration.

Serverless indexes are fully managed by Pinecone and automatically scale
based on usage.

:param cloud: The cloud provider ("aws", "gcp", or "azure").
:param region: The cloud region (e.g., "us-east-1", "us-central1").

Example usage::

from pinecone import ServerlessDeployment

deployment = ServerlessDeployment(cloud="aws", region="us-east-1")

pc.create_index(
name="my-index",
schema=schema,
deployment=deployment,
)
"""

cloud: str
region: str

def to_dict(self) -> dict:
"""Serialize to API format.

:returns: Dictionary representation for the API.
"""
return {"deployment_type": "serverless", "cloud": self.cloud, "region": self.region}


@dataclass
class ByocDeployment:
"""Bring-your-own-cloud (BYOC) deployment configuration.

BYOC deployments run in your own cloud infrastructure with a
Pinecone-managed control plane.

:param environment: The BYOC environment identifier.

Example usage::

from pinecone import ByocDeployment

deployment = ByocDeployment(environment="aws-us-east-1-b92")

pc.create_index(
name="my-index",
schema=schema,
deployment=deployment,
)
"""

environment: str

def to_dict(self) -> dict:
"""Serialize to API format.

:returns: Dictionary representation for the API.
"""
return {"deployment_type": "byoc", "environment": self.environment}


@dataclass
class PodDeployment:
"""Pod-based deployment configuration.

Pod deployments provide dedicated compute resources with configurable
replicas, shards, and pod types.

:param environment: The pod environment (e.g., "us-east-1-aws").
:param pod_type: The pod type (e.g., "p1.x1", "s1.x1", "p2.x1").
:param replicas: Number of replicas (default: 1).
:param shards: Number of shards (default: 1).
:param pods: Total number of pods (replicas * shards). If not specified,
it will be calculated from replicas and shards.

Example usage::

from pinecone import PodDeployment

deployment = PodDeployment(
environment="us-east-1-aws",
pod_type="p1.x1",
replicas=2,
shards=1,
)

pc.create_index(
name="my-index",
schema=schema,
deployment=deployment,
)
"""

environment: str
pod_type: str
replicas: int = 1
shards: int = 1
pods: int | None = None

def to_dict(self) -> dict:
"""Serialize to API format.

:returns: Dictionary representation for the API.
"""
result: dict = {
"deployment_type": "pod",
"environment": self.environment,
"pod_type": self.pod_type,
"replicas": self.replicas,
"shards": self.shards,
}
if self.pods is not None:
result["pods"] = self.pods
return result


# Type alias for any deployment type
Deployment = ServerlessDeployment | ByocDeployment | PodDeployment
149 changes: 149 additions & 0 deletions tests/unit/models/test_deployment.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
"""Tests for deployment model classes."""

import os
import sys
import types


def _load_deployment_module():
"""Load deployment.py as a standalone module to avoid broken imports."""
module_name = "pinecone.db_control.models.deployment"
module = types.ModuleType(module_name)
module.__file__ = os.path.join(
os.path.dirname(__file__),
"..",
"..",
"..",
"pinecone",
"db_control",
"models",
"deployment.py",
)
sys.modules[module_name] = module
with open(module.__file__) as f:
exec(compile(f.read(), module.__file__, "exec"), module.__dict__)
return module


_deployment = _load_deployment_module()
ServerlessDeployment = _deployment.ServerlessDeployment
ByocDeployment = _deployment.ByocDeployment
PodDeployment = _deployment.PodDeployment


class TestServerlessDeployment:
def test_required_params(self):
deployment = ServerlessDeployment(cloud="aws", region="us-east-1")
assert deployment.cloud == "aws"
assert deployment.region == "us-east-1"

def test_to_dict(self):
deployment = ServerlessDeployment(cloud="aws", region="us-east-1")
result = deployment.to_dict()
assert result == {"deployment_type": "serverless", "cloud": "aws", "region": "us-east-1"}

def test_to_dict_gcp(self):
deployment = ServerlessDeployment(cloud="gcp", region="us-central1")
result = deployment.to_dict()
assert result == {"deployment_type": "serverless", "cloud": "gcp", "region": "us-central1"}

def test_to_dict_azure(self):
deployment = ServerlessDeployment(cloud="azure", region="eastus")
result = deployment.to_dict()
assert result == {"deployment_type": "serverless", "cloud": "azure", "region": "eastus"}


class TestByocDeployment:
def test_required_params(self):
deployment = ByocDeployment(environment="aws-us-east-1-b92")
assert deployment.environment == "aws-us-east-1-b92"

def test_to_dict(self):
deployment = ByocDeployment(environment="aws-us-east-1-b92")
result = deployment.to_dict()
assert result == {"deployment_type": "byoc", "environment": "aws-us-east-1-b92"}

def test_to_dict_different_environment(self):
deployment = ByocDeployment(environment="gcp-us-central1-abc")
result = deployment.to_dict()
assert result == {"deployment_type": "byoc", "environment": "gcp-us-central1-abc"}


class TestPodDeployment:
def test_required_params(self):
deployment = PodDeployment(environment="us-east-1-aws", pod_type="p1.x1")
assert deployment.environment == "us-east-1-aws"
assert deployment.pod_type == "p1.x1"
assert deployment.replicas == 1
assert deployment.shards == 1
assert deployment.pods is None

def test_to_dict_minimal(self):
deployment = PodDeployment(environment="us-east-1-aws", pod_type="p1.x1")
result = deployment.to_dict()
assert result == {
"deployment_type": "pod",
"environment": "us-east-1-aws",
"pod_type": "p1.x1",
"replicas": 1,
"shards": 1,
}

def test_to_dict_with_replicas(self):
deployment = PodDeployment(environment="us-east-1-aws", pod_type="p1.x1", replicas=3)
result = deployment.to_dict()
assert result == {
"deployment_type": "pod",
"environment": "us-east-1-aws",
"pod_type": "p1.x1",
"replicas": 3,
"shards": 1,
}

def test_to_dict_with_shards(self):
deployment = PodDeployment(environment="us-east-1-aws", pod_type="p1.x1", shards=2)
result = deployment.to_dict()
assert result == {
"deployment_type": "pod",
"environment": "us-east-1-aws",
"pod_type": "p1.x1",
"replicas": 1,
"shards": 2,
}

def test_to_dict_with_pods(self):
deployment = PodDeployment(
environment="us-east-1-aws", pod_type="p1.x1", replicas=2, shards=2, pods=4
)
result = deployment.to_dict()
assert result == {
"deployment_type": "pod",
"environment": "us-east-1-aws",
"pod_type": "p1.x1",
"replicas": 2,
"shards": 2,
"pods": 4,
}

def test_to_dict_different_pod_types(self):
for pod_type in ["s1.x1", "s1.x2", "p1.x2", "p2.x1"]:
deployment = PodDeployment(environment="us-east-1-aws", pod_type=pod_type)
result = deployment.to_dict()
assert result["pod_type"] == pod_type


class TestDeploymentUsageExamples:
"""Test the usage examples from the ticket."""

def test_serverless_deployment_example(self):
deployment = ServerlessDeployment(cloud="gcp", region="us-central1")
result = deployment.to_dict()
assert result["deployment_type"] == "serverless"
assert result["cloud"] == "gcp"
assert result["region"] == "us-central1"

def test_byoc_deployment_example(self):
deployment = ByocDeployment(environment="aws-us-east-1-b92")
result = deployment.to_dict()
assert result["deployment_type"] == "byoc"
assert result["environment"] == "aws-us-east-1-b92"