diff --git a/pinecone/__init__.py b/pinecone/__init__.py index 1600db72..c1b0f6ac 100644 --- a/pinecone/__init__.py +++ b/pinecone/__init__.py @@ -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", diff --git a/pinecone/db_control/models/__init__.py b/pinecone/db_control/models/__init__.py index 9e5338b3..eb958414 100644 --- a/pinecone/db_control/models/__init__.py +++ b/pinecone/db_control/models/__init__.py @@ -21,6 +21,7 @@ SchemaField, ) from .schema_builder import SchemaBuilder +from .deployment import ServerlessDeployment, ByocDeployment, PodDeployment, Deployment __all__ = [ @@ -46,4 +47,8 @@ "SemanticTextField", "SchemaField", "SchemaBuilder", + "ServerlessDeployment", + "ByocDeployment", + "PodDeployment", + "Deployment", ] diff --git a/pinecone/db_control/models/deployment.py b/pinecone/db_control/models/deployment.py new file mode 100644 index 00000000..81e61cb0 --- /dev/null +++ b/pinecone/db_control/models/deployment.py @@ -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 diff --git a/tests/unit/models/test_deployment.py b/tests/unit/models/test_deployment.py new file mode 100644 index 00000000..0c8c50f8 --- /dev/null +++ b/tests/unit/models/test_deployment.py @@ -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"