diff --git a/pinecone/db_control/request_factory.py b/pinecone/db_control/request_factory.py
index c6738215a..dd05ccca0 100644
--- a/pinecone/db_control/request_factory.py
+++ b/pinecone/db_control/request_factory.py
@@ -15,39 +15,30 @@
 )
 from pinecone.core.openapi.db_control.model.create_index_request import CreateIndexRequest
 from pinecone.core.openapi.db_control.model.configure_index_request import ConfigureIndexRequest
-from pinecone.core.openapi.db_control.model.configure_index_request_embed import (
-    ConfigureIndexRequestEmbed,
-)
-from pinecone.core.openapi.db_control.model.index_spec import IndexSpec
 from pinecone.core.openapi.db_control.model.index_tags import IndexTags
-from pinecone.core.openapi.db_control.model.serverless_spec import (
-    ServerlessSpec as ServerlessSpecModel,
-)
 from pinecone.core.openapi.db_control.model.read_capacity_on_demand_spec import (
     ReadCapacityOnDemandSpec,
 )
 from pinecone.core.openapi.db_control.model.read_capacity_dedicated_spec import (
     ReadCapacityDedicatedSpec,
 )
-from pinecone.core.openapi.db_control.model.read_capacity_dedicated_config import (
-    ReadCapacityDedicatedConfig,
-)
-from pinecone.core.openapi.db_control.model.scaling_config_manual import ScalingConfigManual
-from pinecone.core.openapi.db_control.model.backup_model_schema import BackupModelSchema
-from pinecone.core.openapi.db_control.model.backup_model_schema_fields import (
-    BackupModelSchemaFields,
+from pinecone.core.openapi.db_control.model.schema import Schema
+from pinecone.core.openapi.db_control.model.schema_fields import SchemaFields
+from pinecone.core.openapi.db_control.model.pod_deployment_metadata_config import (
+    PodDeploymentMetadataConfig,
 )
-from pinecone.core.openapi.db_control.model.byoc_spec import ByocSpec as ByocSpecModel
-from pinecone.core.openapi.db_control.model.pod_spec import PodSpec as PodSpecModel
-from pinecone.core.openapi.db_control.model.pod_spec_metadata_config import PodSpecMetadataConfig
 from pinecone.core.openapi.db_control.model.create_index_from_backup_request import (
     CreateIndexFromBackupRequest,
 )
 from pinecone.db_control.models import ServerlessSpec, PodSpec, ByocSpec, IndexModel, IndexEmbed
-from pinecone.db_control.models.deployment import (
-    ServerlessDeployment,
-    PodDeployment,
-    ByocDeployment,
+from pinecone.core.openapi.db_control.model.serverless_deployment import (
+    ServerlessDeployment as ServerlessDeploymentModel,
+)
+from pinecone.core.openapi.db_control.model.pod_deployment import (
+    PodDeployment as PodDeploymentModel,
+)
+from pinecone.core.openapi.db_control.model.byoc_deployment import (
+    ByocDeployment as ByocDeploymentModel,
 )
 from pinecone.db_control.models.schema_fields import DenseVectorField, SparseVectorField
 
@@ -61,7 +52,7 @@
     GcpRegion,
     AzureRegion,
 )
-from .types import CreateIndexForModelEmbedTypedDict, ConfigureIndexEmbed
+from .types import CreateIndexForModelEmbedTypedDict
 
 if TYPE_CHECKING:
     from pinecone.db_control.models.serverless_spec import (
@@ -69,6 +60,11 @@
         MetadataSchemaFieldConfig,
     )
     from pinecone.core.openapi.db_control.model.read_capacity import ReadCapacity
+    from pinecone.db_control.models.deployment import (
+        ServerlessDeployment,
+        PodDeployment,
+        ByocDeployment,
+    )
 
 logger = logging.getLogger(__name__)
 """ :meta private: """
@@ -122,9 +118,9 @@ def __parse_read_capacity(
                 result = ReadCapacityOnDemandSpec(mode="OnDemand")
                 return cast(ReadCapacityOnDemandSpec, result)
             elif mode == "Dedicated":
+                # New API structure: ReadCapacityDedicatedSpec has node_type, scaling (object with strategy, replicas, shards)
                 dedicated_dict: dict[str, Any] = read_capacity.get("dedicated", {})  # type: ignore
-                # Construct ReadCapacityDedicatedConfig
-                # node_type and scaling are required fields
+
                 if "node_type" not in dedicated_dict or dedicated_dict.get("node_type") is None:
                     raise ValueError(
                         "node_type is required when using Dedicated read capacity mode. "
@@ -135,42 +131,56 @@ def __parse_read_capacity(
                         "scaling is required when using Dedicated read capacity mode. "
                         "Please specify 'scaling' (e.g., 'Manual') in the 'dedicated' configuration."
                     )
+
                 node_type = dedicated_dict["node_type"]
-                scaling = dedicated_dict["scaling"]
-                dedicated_config_kwargs = {"node_type": node_type, "scaling": scaling}
-
-                # Validate that manual scaling configuration is provided when scaling is "Manual"
-                if scaling == "Manual":
-                    if "manual" not in dedicated_dict or dedicated_dict.get("manual") is None:
-                        raise ValueError(
-                            "When using 'Manual' scaling with Dedicated read capacity mode, "
-                            "the 'manual' field with 'shards' and 'replicas' is required. "
-                            "Please specify 'manual': {'shards': <number>, 'replicas': <number>} "
-                            "in the 'dedicated' configuration."
-                        )
-                    manual_dict = dedicated_dict["manual"]
-                    if not isinstance(manual_dict, dict):
-                        raise ValueError(
-                            "The 'manual' field must be a dictionary with 'shards' and 'replicas' keys."
+                scaling_value = dedicated_dict["scaling"]
+
+                # Handle manual scaling configuration
+                from pinecone.core.openapi.db_control.model.read_capacity_dedicated_spec_response_scaling import (
+                    ReadCapacityDedicatedSpecResponseScaling,
+                )
+
+                if isinstance(scaling_value, str):
+                    # Old format: scaling is a string like "Manual", manual dict has shards/replicas
+                    if scaling_value == "Manual":
+                        if "manual" not in dedicated_dict or dedicated_dict.get("manual") is None:
+                            raise ValueError(
+                                "When using 'Manual' scaling with Dedicated read capacity mode, "
+                                "the 'manual' field with 'shards' and 'replicas' is required."
+                            )
+                        manual_dict = dedicated_dict["manual"]
+                        if not isinstance(manual_dict, dict):
+                            raise ValueError(
+                                "The 'manual' field must be a dictionary with 'shards' and 'replicas' keys."
+                            )
+                        if "shards" not in manual_dict or "replicas" not in manual_dict:
+                            missing = []
+                            if "shards" not in manual_dict:
+                                missing.append("shards")
+                            if "replicas" not in manual_dict:
+                                missing.append("replicas")
+                            raise ValueError(
+                                f"The 'manual' configuration is missing required fields: {', '.join(missing)}."
+                            )
+
+                        # Create scaling object with strategy, replicas, shards
+                        scaling_obj = ReadCapacityDedicatedSpecResponseScaling(
+                            strategy=scaling_value,
+                            replicas=manual_dict["replicas"],
+                            shards=manual_dict["shards"],
                         )
-                    if "shards" not in manual_dict or "replicas" not in manual_dict:
-                        missing = []
-                        if "shards" not in manual_dict:
-                            missing.append("shards")
-                        if "replicas" not in manual_dict:
-                            missing.append("replicas")
-                        raise ValueError(
-                            f"The 'manual' configuration is missing required fields: {', '.join(missing)}. "
-                            "Please provide both 'shards' and 'replicas' in the 'manual' configuration."
+                    else:
+                        # Other scaling strategies might not need manual config
+                        scaling_obj = ReadCapacityDedicatedSpecResponseScaling(
+                            strategy=scaling_value, replicas=1, shards=1
                         )
-                    dedicated_config_kwargs["manual"] = ScalingConfigManual(**manual_dict)
-                elif "manual" in dedicated_dict:
-                    # Allow manual to be provided for other scaling types (future compatibility)
-                    manual_dict = dedicated_dict["manual"]
-                    dedicated_config_kwargs["manual"] = ScalingConfigManual(**manual_dict)
-
-                dedicated_config = ReadCapacityDedicatedConfig(**dedicated_config_kwargs)
-                result = ReadCapacityDedicatedSpec(mode="Dedicated", dedicated=dedicated_config)
+                else:
+                    # Already a scaling object
+                    scaling_obj = scaling_value
+
+                result = ReadCapacityDedicatedSpec(
+                    mode="Dedicated", node_type=node_type, scaling=scaling_obj
+                )
                 return cast(ReadCapacityDedicatedSpec, result)
             else:
                 # Fallback: let OpenAPI handle it
@@ -193,14 +203,14 @@ def __parse_schema(
             | dict[
                 str, dict[str, Any]
             ]  # Dict with "fields" wrapper: {"fields": {field_name: {...}}, ...}
-            | BackupModelSchema  # OpenAPI model instance
+            | Schema  # OpenAPI model instance
         ),
-    ) -> BackupModelSchema:
-        """Parse schema dict into BackupModelSchema instance.
+    ) -> Schema:
+        """Parse schema dict into Schema instance.
 
         :param schema: Dict with schema configuration (either {field_name: {filterable: bool, ...}} or
-            {"fields": {field_name: {filterable: bool, ...}}, ...}) or existing BackupModelSchema instance
-        :return: BackupModelSchema instance
+            {"fields": {field_name: {filterable: bool, ...}}, ...}) or existing Schema instance
+        :return: Schema instance
         """
         if isinstance(schema, dict):
             schema_kwargs: dict[str, Any] = {}
@@ -213,10 +223,10 @@ def __parse_schema(
                 for field_name, field_config in schema["fields"].items():
                     if isinstance(field_config, dict):
                         # Pass through the entire field_config dict to allow future API fields
-                        fields[field_name] = BackupModelSchemaFields(**field_config)
+                        fields[field_name] = SchemaFields(**field_config)
                     else:
                         # If not a dict, create with default filterable=True
-                        fields[field_name] = BackupModelSchemaFields(filterable=True)
+                        fields[field_name] = SchemaFields(filterable=True)
                 schema_kwargs["fields"] = fields
 
                 # Pass through any other fields in schema_dict to allow future API fields
@@ -230,14 +240,14 @@ def __parse_schema(
                 for field_name, field_config in schema.items():
                     if isinstance(field_config, dict):
                         # Pass through the entire field_config dict to allow future API fields
-                        fields[field_name] = BackupModelSchemaFields(**field_config)
+                        fields[field_name] = SchemaFields(**field_config)
                     else:
                         # If not a dict, create with default filterable=True
-                        fields[field_name] = BackupModelSchemaFields(filterable=True)
+                        fields[field_name] = SchemaFields(filterable=True)
                 # Ensure fields is always set, even if empty
                 schema_kwargs["fields"] = fields
 
-            # Validate that fields is present before constructing BackupModelSchema
+            # Validate that fields is present before constructing Schema
             if "fields" not in schema_kwargs:
                 raise ValueError(
                     "Schema dict must contain field definitions. "
@@ -247,126 +257,12 @@ def __parse_schema(
 
             from typing import cast
 
-            result = BackupModelSchema(**schema_kwargs)
-            return cast(BackupModelSchema, result)
+            result = Schema(**schema_kwargs)
+            return cast(Schema, result)
         else:
-            # Already a BackupModelSchema instance
+            # Already a Schema instance
             return schema
 
-    @staticmethod
-    def __parse_index_spec(spec: Dict | ServerlessSpec | PodSpec | ByocSpec) -> IndexSpec:
-        if isinstance(spec, dict):
-            if "serverless" in spec:
-                spec["serverless"]["cloud"] = convert_enum_to_string(spec["serverless"]["cloud"])
-                spec["serverless"]["region"] = convert_enum_to_string(spec["serverless"]["region"])
-
-                # Handle read_capacity if present
-                if "read_capacity" in spec["serverless"]:
-                    spec["serverless"]["read_capacity"] = (
-                        PineconeDBControlRequestFactory.__parse_read_capacity(
-                            spec["serverless"]["read_capacity"]
-                        )
-                    )
-
-                # Handle schema if present - convert to BackupModelSchema
-                if "schema" in spec["serverless"]:
-                    schema_dict = spec["serverless"]["schema"]
-                    if isinstance(schema_dict, dict):
-                        # Process fields if present, otherwise pass through as-is
-                        schema_kwargs = {}
-                        if "fields" in schema_dict:
-                            fields = {}
-                            for field_name, field_config in schema_dict["fields"].items():
-                                if isinstance(field_config, dict):
-                                    # Pass through the entire field_config dict to allow future API fields
-                                    fields[field_name] = BackupModelSchemaFields(**field_config)
-                                else:
-                                    # If not a dict, create with default filterable=True
-                                    fields[field_name] = BackupModelSchemaFields(filterable=True)
-                            schema_kwargs["fields"] = fields
-
-                        # Pass through any other fields in schema_dict to allow future API fields
-                        for key, value in schema_dict.items():
-                            if key != "fields":
-                                schema_kwargs[key] = value
-
-                        spec["serverless"]["schema"] = BackupModelSchema(**schema_kwargs)
-
-                index_spec = IndexSpec(serverless=ServerlessSpecModel(**spec["serverless"]))
-            elif "pod" in spec:
-                spec["pod"]["environment"] = convert_enum_to_string(spec["pod"]["environment"])
-                args_dict = parse_non_empty_args(
-                    [
-                        ("environment", spec["pod"].get("environment")),
-                        ("metadata_config", spec["pod"].get("metadata_config")),
-                        ("replicas", spec["pod"].get("replicas")),
-                        ("shards", spec["pod"].get("shards")),
-                        ("pods", spec["pod"].get("pods")),
-                        ("source_collection", spec["pod"].get("source_collection")),
-                    ]
-                )
-                if args_dict.get("metadata_config"):
-                    args_dict["metadata_config"] = PodSpecMetadataConfig(
-                        indexed=args_dict["metadata_config"].get("indexed", None)
-                    )
-                index_spec = IndexSpec(pod=PodSpecModel(**args_dict))
-            elif "byoc" in spec:
-                index_spec = IndexSpec(byoc=ByocSpecModel(**spec["byoc"]))
-            else:
-                raise ValueError("spec must contain either 'serverless', 'pod', or 'byoc' key")
-        elif isinstance(spec, ServerlessSpec):
-            # Build args dict for ServerlessSpecModel
-            serverless_args: dict[str, Any] = {"cloud": spec.cloud, "region": spec.region}
-
-            # Handle read_capacity
-            if spec.read_capacity is not None:
-                serverless_args["read_capacity"] = (
-                    PineconeDBControlRequestFactory.__parse_read_capacity(spec.read_capacity)
-                )
-
-            # Handle schema
-            if spec.schema is not None:
-                # Convert dict to BackupModelSchema
-                # schema is {field_name: {filterable: bool, ...}}
-                # Pass through the entire field_config to allow future API fields
-                fields = {}
-                for field_name, field_config in spec.schema.items():
-                    if isinstance(field_config, dict):
-                        # Pass through the entire field_config dict to allow future API fields
-                        fields[field_name] = BackupModelSchemaFields(**field_config)
-                    else:
-                        # If not a dict, create with default filterable=True
-                        fields[field_name] = BackupModelSchemaFields(filterable=True)
-                serverless_args["schema"] = BackupModelSchema(fields=fields)
-
-            index_spec = IndexSpec(serverless=ServerlessSpecModel(**serverless_args))
-        elif isinstance(spec, PodSpec):
-            args_dict = parse_non_empty_args(
-                [
-                    ("replicas", spec.replicas),
-                    ("shards", spec.shards),
-                    ("pods", spec.pods),
-                    ("source_collection", spec.source_collection),
-                ]
-            )
-            if spec.metadata_config:
-                args_dict["metadata_config"] = PodSpecMetadataConfig(
-                    indexed=spec.metadata_config.get("indexed", None)
-                )
-
-            index_spec = IndexSpec(
-                pod=PodSpecModel(environment=spec.environment, pod_type=spec.pod_type, **args_dict)
-            )
-        elif isinstance(spec, ByocSpec):
-            args_dict = parse_non_empty_args([("environment", spec.environment)])
-            index_spec = IndexSpec(byoc=ByocSpecModel(**args_dict))
-        else:
-            raise TypeError("spec must be of type dict, ServerlessSpec, PodSpec, or ByocSpec")
-
-        from typing import cast
-
-        return cast(IndexSpec, index_spec)
-
     @staticmethod
     def _translate_legacy_request(
         spec: Dict | ServerlessSpec | PodSpec | ByocSpec,
@@ -415,37 +311,60 @@ def _translate_legacy_request(
 
         # Translate spec to deployment
         deployment_dict: dict[str, Any]
+        deployment: ServerlessDeploymentModel | PodDeploymentModel | ByocDeploymentModel
         if isinstance(spec, dict):
             if "serverless" in spec:
                 serverless_spec = spec["serverless"]
                 # Convert enum values to strings for consistency with __parse_index_spec
                 cloud = convert_enum_to_string(serverless_spec.get("cloud", ""))
                 region = convert_enum_to_string(serverless_spec.get("region", ""))
-                deployment = ServerlessDeployment(cloud=cloud, region=region)
+                deployment = ServerlessDeploymentModel(
+                    deployment_type="serverless", cloud=cloud, region=region
+                )
                 deployment_dict = deployment.to_dict()
             elif "pod" in spec:
                 pod_spec = spec["pod"]
                 # Convert enum values to strings for consistency with __parse_index_spec
                 environment = convert_enum_to_string(pod_spec.get("environment", ""))
                 pod_type = convert_enum_to_string(pod_spec.get("pod_type", "p1.x1"))
-                deployment = PodDeployment(
-                    environment=environment,
-                    pod_type=pod_type,
-                    replicas=pod_spec.get("replicas", 1),
-                    shards=pod_spec.get("shards", 1),
-                    pods=pod_spec.get("pods"),
-                )
+                pod_kwargs: dict[str, Any] = {
+                    "deployment_type": "pod",
+                    "environment": environment,
+                    "pod_type": pod_type,
+                    "replicas": pod_spec.get("replicas")
+                    if pod_spec.get("replicas") is not None
+                    else 1,
+                    "shards": pod_spec.get("shards") if pod_spec.get("shards") is not None else 1,
+                }
+                # Only include pods if it's explicitly provided and not None
+                pods_value = pod_spec.get("pods")
+                if pods_value is not None:
+                    pod_kwargs["pods"] = pods_value
+                # Handle source_collection if present
+                source_collection_value = pod_spec.get("source_collection")
+                if source_collection_value is not None:
+                    pod_kwargs["source_collection"] = source_collection_value
+                # Handle metadata_config if present - convert to PodDeploymentMetadataConfig
+                if pod_spec.get("metadata_config"):
+                    metadata_config_dict = pod_spec["metadata_config"]
+                    if isinstance(metadata_config_dict, dict) and metadata_config_dict:
+                        pod_kwargs["metadata_config"] = PodDeploymentMetadataConfig(
+                            **metadata_config_dict
+                        )
+                deployment = PodDeploymentModel(**pod_kwargs)
                 deployment_dict = deployment.to_dict()
             elif "byoc" in spec:
                 byoc_spec = spec["byoc"]
                 # Convert enum values to strings for consistency with __parse_index_spec
                 environment = convert_enum_to_string(byoc_spec.get("environment", ""))
-                deployment = ByocDeployment(environment=environment)
+                deployment = ByocDeploymentModel(deployment_type="byoc", environment=environment)
                 deployment_dict = deployment.to_dict()
             else:
                 raise ValueError("spec must contain either 'serverless', 'pod', or 'byoc' key")
         elif isinstance(spec, ServerlessSpec):
-            deployment = ServerlessDeployment(cloud=spec.cloud, region=spec.region)
+            deployment = ServerlessDeploymentModel(
+                deployment_type="serverless", cloud=spec.cloud, region=spec.region
+            )
             deployment_dict = deployment.to_dict()
         elif isinstance(spec, PodSpec):
             # PodDeployment requires pod_type, but PodSpec defaults to "p1.x1"
@@ -453,16 +372,28 @@ def _translate_legacy_request(
             # Use explicit None check to preserve 0 values (consistent with dict handling)
             replicas = spec.replicas if spec.replicas is not None else 1
             shards = spec.shards if spec.shards is not None else 1
-            deployment = PodDeployment(
-                environment=spec.environment,
-                pod_type=pod_type,
-                replicas=replicas,
-                shards=shards,
-                pods=spec.pods,
-            )
+            pod_obj_kwargs: dict[str, Any] = {
+                "deployment_type": "pod",
+                "environment": spec.environment,
+                "pod_type": pod_type,
+                "replicas": replicas,
+                "shards": shards,
+            }
+            # Only include pods if it's not None
+            if spec.pods is not None:
+                pod_obj_kwargs["pods"] = spec.pods
+            # Handle source_collection if present
+            if spec.source_collection is not None:
+                pod_obj_kwargs["source_collection"] = spec.source_collection
+            # Handle metadata_config if present - convert to PodDeploymentMetadataConfig
+            if spec.metadata_config:
+                pod_obj_kwargs["metadata_config"] = PodDeploymentMetadataConfig(
+                    **spec.metadata_config
+                )
+            deployment = PodDeploymentModel(**pod_obj_kwargs)
             deployment_dict = deployment.to_dict()
         elif isinstance(spec, ByocSpec):
-            deployment = ByocDeployment(environment=spec.environment)
+            deployment = ByocDeploymentModel(deployment_type="byoc", environment=spec.environment)
             deployment_dict = deployment.to_dict()
         else:
             raise TypeError("spec must be of type dict, ServerlessSpec, PodSpec, or ByocSpec")
@@ -533,7 +464,9 @@ def _translate_embed_to_semantic_text(
         region = convert_enum_to_string(region)
 
         # Create ServerlessDeployment for cloud/region
-        deployment = ServerlessDeployment(cloud=cloud, region=region)
+        deployment = ServerlessDeploymentModel(
+            deployment_type="serverless", cloud=cloud, region=region
+        )
         deployment_dict = deployment.to_dict()
 
         # Parse embed configuration
@@ -624,7 +557,7 @@ def _serialize_schema(schema: dict[str, Any]) -> dict[str, Any]:
     def create_index_with_schema_request(
         name: str,
         schema: dict[str, Any],
-        deployment: ServerlessDeployment | PodDeployment | ByocDeployment | None = None,
+        deployment: "ServerlessDeployment | PodDeployment | ByocDeployment | None" = None,
         deletion_protection: (DeletionProtection | str) | None = DeletionProtection.DISABLED,
         tags: dict[str, str] | None = None,
     ) -> CreateIndexRequest:
@@ -649,16 +582,31 @@ def create_index_with_schema_request(
 
         # Default deployment to aws/us-east-1 serverless
         if deployment is None:
-            deployment = ServerlessDeployment(cloud="aws", region="us-east-1")
+            from pinecone.db_control.models.deployment import (
+                ServerlessDeployment as SDKServerlessDeployment,
+            )
+
+            deployment = SDKServerlessDeployment(cloud="aws", region="us-east-1")
 
+        # Convert SDK deployment to dict, then let CreateIndexRequest deserialize it
         deployment_dict = deployment.to_dict()
+
+        # Import the Deployment union class and use it to properly deserialize
+        from pinecone.core.openapi.db_control.model.deployment import Deployment
+
+        parsed_deployment = Deployment(**deployment_dict)
+
+        # Parse schema
         schema_dict = PineconeDBControlRequestFactory._serialize_schema(schema)
+        parsed_schema = None
+        if schema_dict and schema_dict.get("fields"):
+            parsed_schema = PineconeDBControlRequestFactory.__parse_schema(schema_dict)
 
         args = parse_non_empty_args(
             [
                 ("name", name),
-                ("deployment", deployment_dict),
-                ("schema", schema_dict),
+                ("deployment", parsed_deployment),
+                ("schema", parsed_schema),
                 ("deletion_protection", dp),
                 ("tags", tags_obj),
             ]
@@ -679,6 +627,11 @@ def create_index_request(
         vector_type: (VectorType | str) | None = VectorType.DENSE,
         tags: dict[str, str] | None = None,
     ) -> CreateIndexRequest:
+        """Create a CreateIndexRequest using legacy parameters.
+
+        This method translates legacy spec/dimension/metric parameters to the new
+        deployment/schema format required by the current API.
+        """
         if metric is not None:
             metric = convert_enum_to_string(metric)
         if vector_type is not None:
@@ -689,19 +642,39 @@ def create_index_request(
             dp = None
 
         tags_obj = PineconeDBControlRequestFactory.__parse_tags(tags)
-        index_spec = PineconeDBControlRequestFactory.__parse_index_spec(spec)
 
         if vector_type == VectorType.SPARSE.value and dimension is not None:
             raise ValueError("dimension should not be specified for sparse indexes")
 
+        # Translate legacy parameters to new deployment + schema format
+        deployment_dict, schema_dict = PineconeDBControlRequestFactory._translate_legacy_request(
+            spec=spec, dimension=dimension, metric=metric, vector_type=vector_type
+        )
+
+        # Convert metadata_config dict to PodDeploymentMetadataConfig if present
+        if deployment_dict.get("deployment_type") == "pod" and "metadata_config" in deployment_dict:
+            metadata_config_dict = deployment_dict["metadata_config"]
+            if isinstance(metadata_config_dict, dict):
+                deployment_dict["metadata_config"] = PodDeploymentMetadataConfig(
+                    **metadata_config_dict
+                )
+
+        # Convert deployment_dict to Deployment object using the discriminator
+        from pinecone.core.openapi.db_control.model.deployment import Deployment
+
+        parsed_deployment = Deployment(**deployment_dict)
+
+        # Convert schema_dict to Schema object if present
+        parsed_schema = None
+        if schema_dict and schema_dict.get("fields"):
+            parsed_schema = PineconeDBControlRequestFactory.__parse_schema(schema_dict)
+
         args = parse_non_empty_args(
             [
                 ("name", name),
-                ("dimension", dimension),
-                ("metric", metric),
-                ("spec", index_spec),
+                ("deployment", parsed_deployment),
+                ("schema", parsed_schema),
                 ("deletion_protection", dp),
-                ("vector_type", vector_type),
                 ("tags", tags_obj),
             ]
         )
@@ -733,7 +706,7 @@ def create_index_for_model_request(
             | dict[
                 str, dict[str, Any]
             ]  # Dict with "fields" wrapper: {"fields": {field_name: {...}}, ...}
-            | BackupModelSchema  # OpenAPI model instance
+            | Schema  # OpenAPI model instance
         )
         | None = None,
     ) -> CreateIndexForModelRequest:
@@ -816,7 +789,6 @@ def configure_index_request(
         pod_type: (PodType | str) | None = None,
         deletion_protection: (DeletionProtection | str) | None = None,
         tags: dict[str, str] | None = None,
-        embed: (ConfigureIndexEmbed | Dict) | None = None,
         read_capacity: (
             "ReadCapacityDict"
             | "ReadCapacity"
@@ -835,10 +807,11 @@ def configure_index_request(
             raise ValueError("deletion_protection must be either 'enabled' or 'disabled'")
 
         fetched_tags = description.tags
+        starting_tags: dict[str, str]
         if fetched_tags is None:
             starting_tags = {}
         else:
-            starting_tags = fetched_tags.to_dict()
+            starting_tags = fetched_tags.to_dict() if hasattr(fetched_tags, "to_dict") else {}
 
         if tags is None:
             # Do not modify tags if none are provided
@@ -854,10 +827,6 @@ def configure_index_request(
         if replicas:
             pod_config_args.update(replicas=replicas)
 
-        embed_config = None
-        if embed is not None:
-            embed_config = ConfigureIndexRequestEmbed(**dict(embed))
-
         # Parse read_capacity if provided
         parsed_read_capacity = None
         if read_capacity is not None:
@@ -873,12 +842,7 @@ def configure_index_request(
             spec = {"serverless": {"read_capacity": parsed_read_capacity}}
 
         args_dict = parse_non_empty_args(
-            [
-                ("deletion_protection", dp),
-                ("tags", IndexTags(**tags)),
-                ("spec", spec),
-                ("embed", embed_config),
-            ]
+            [("deletion_protection", dp), ("tags", IndexTags(**tags)), ("spec", spec)]
         )
 
         from typing import cast
diff --git a/pinecone/db_control/resources/asyncio/index.py b/pinecone/db_control/resources/asyncio/index.py
index 8b993183a..c243cdbd6 100644
--- a/pinecone/db_control/resources/asyncio/index.py
+++ b/pinecone/db_control/resources/asyncio/index.py
@@ -32,7 +32,6 @@
 from pinecone.db_control.request_factory import PineconeDBControlRequestFactory
 from pinecone.core.openapi.db_control import API_VERSION
 from pinecone.utils import require_kwargs
-from pinecone.db_control.types.configure_index_embed import ConfigureIndexEmbed
 
 logger = logging.getLogger(__name__)
 """ :meta private: """
@@ -43,13 +42,13 @@
         MetadataSchemaFieldConfig,
     )
     from pinecone.core.openapi.db_control.model.read_capacity import ReadCapacity
+    from pinecone.core.openapi.db_control.model.schema import Schema
     from pinecone.core.openapi.db_control.model.read_capacity_on_demand_spec import (
         ReadCapacityOnDemandSpec,
     )
     from pinecone.core.openapi.db_control.model.read_capacity_dedicated_spec import (
         ReadCapacityDedicatedSpec,
     )
-    from pinecone.core.openapi.db_control.model.backup_model_schema import BackupModelSchema
 
 
 class IndexResourceAsyncio:
@@ -137,6 +136,8 @@ async def create(
             )
         else:
             # Legacy spec-based creation
+            # spec is guaranteed to be non-None here due to validation above
+            assert spec is not None, "spec must be provided when schema is not specified"
             req = PineconeDBControlRequestFactory.create_index_request(
                 name=name,
                 spec=spec,
@@ -179,7 +180,7 @@ async def create_for_model(
             | dict[
                 str, dict[str, Any]
             ]  # Dict with "fields" wrapper: {"fields": {field_name: {...}}, ...}
-            | "BackupModelSchema"  # OpenAPI model instance
+            | "Schema"  # OpenAPI model instance
         )
         | None = None,
         timeout: int | None = None,
@@ -226,9 +227,10 @@ async def __poll_describe_index_until_ready(
         total_wait_time = 0
         while True:
             description = await self.describe(name=name)
-            if description.status.state == "InitializationFailed":
+            status = description.status
+            if hasattr(status, "state") and status.state == "InitializationFailed":
                 raise Exception(f"Index {name} failed to initialize.")
-            if description.status.ready:
+            if hasattr(status, "ready") and status.ready:
                 return description
 
             if timeout is not None and total_wait_time >= timeout:
@@ -296,7 +298,6 @@ async def configure(
         pod_type: (PodType | str) | None = None,
         deletion_protection: (DeletionProtection | str) | None = None,
         tags: dict[str, str] | None = None,
-        embed: (ConfigureIndexEmbed | Dict) | None = None,
         read_capacity: (
             "ReadCapacityDict"
             | "ReadCapacity"
@@ -313,7 +314,6 @@ async def configure(
             pod_type=pod_type,
             deletion_protection=deletion_protection,
             tags=tags,
-            embed=embed,
             read_capacity=read_capacity,
         )
         await self._index_api.configure_index(name, configure_index_request=req)
diff --git a/pinecone/db_control/resources/sync/index.py b/pinecone/db_control/resources/sync/index.py
index 6b3eb86a3..9f5636f68 100644
--- a/pinecone/db_control/resources/sync/index.py
+++ b/pinecone/db_control/resources/sync/index.py
@@ -19,7 +19,6 @@
 from pinecone.db_control.types import CreateIndexForModelEmbedTypedDict
 from pinecone.db_control.request_factory import PineconeDBControlRequestFactory
 from pinecone.core.openapi.db_control import API_VERSION
-from pinecone.db_control.types.configure_index_embed import ConfigureIndexEmbed
 
 logger = logging.getLogger(__name__)
 """ :meta private: """
@@ -27,6 +26,7 @@
 if TYPE_CHECKING:
     from pinecone.config import Config, OpenApiConfiguration
     from pinecone.core.openapi.db_control.api.manage_indexes_api import ManageIndexesApi
+    from pinecone.core.openapi.db_control.model.schema import Schema
     from pinecone.db_control.enums import (
         Metric,
         VectorType,
@@ -49,7 +49,6 @@
     from pinecone.core.openapi.db_control.model.read_capacity_dedicated_spec import (
         ReadCapacityDedicatedSpec,
     )
-    from pinecone.core.openapi.db_control.model.backup_model_schema import BackupModelSchema
 
 
 class IndexResource(PluginAware):
@@ -157,6 +156,8 @@ def create(
             )
         else:
             # Legacy spec-based creation
+            # spec is guaranteed to be non-None here due to validation above
+            assert spec is not None, "spec must be provided when schema is not specified"
             req = PineconeDBControlRequestFactory.create_index_request(
                 name=name,
                 spec=spec,
@@ -199,7 +200,7 @@ def create_for_model(
             | dict[
                 str, dict[str, Any]
             ]  # Dict with "fields" wrapper: {"fields": {field_name: {...}}, ...}
-            | "BackupModelSchema"  # OpenAPI model instance
+            | "Schema"  # OpenAPI model instance
         )
         | None = None,
         timeout: int | None = None,
@@ -263,11 +264,12 @@ def __poll_describe_index_until_ready(
         total_wait_time = 0
         while True:
             description = self.describe(name=name)
-            if description.status.state == "InitializationFailed":
+            status = description.status
+            if hasattr(status, "state") and status.state == "InitializationFailed":
                 raise Exception(
-                    f"Index {name} failed to initialize. The index status is {description.status.state}."
+                    f"Index {name} failed to initialize. The index status is {status.state}."
                 )
-            if description.status.ready:
+            if hasattr(status, "ready") and status.ready:
                 return description
 
             if timeout is not None and total_wait_time >= timeout:
@@ -339,7 +341,6 @@ def configure(
         pod_type: ("PodType" | str) | None = None,
         deletion_protection: ("DeletionProtection" | str) | None = None,
         tags: dict[str, str] | None = None,
-        embed: ("ConfigureIndexEmbed" | Dict) | None = None,
         read_capacity: (
             "ReadCapacityDict"
             | "ReadCapacity"
@@ -357,7 +358,6 @@ def configure(
             pod_type=pod_type,
             deletion_protection=deletion_protection,
             tags=tags,
-            embed=embed,
             read_capacity=read_capacity,
         )
         api_instance.configure_index(name, configure_index_request=req)
diff --git a/tests/fixtures.py b/tests/fixtures.py
new file mode 100644
index 000000000..27297d135
--- /dev/null
+++ b/tests/fixtures.py
@@ -0,0 +1,161 @@
+"""Test fixtures for creating test instances of SDK models.
+
+This module provides helper functions for creating test instances with both
+old-style and new-style API parameters for backward compatibility.
+"""
+
+from typing import Any
+from pinecone.core.openapi.db_control.model.index_model import IndexModel as OpenAPIIndexModel
+from pinecone.core.openapi.db_control.model.schema import Schema
+from pinecone.core.openapi.db_control.model.schema_fields import SchemaFields
+from pinecone.core.openapi.db_control.model.serverless_deployment import ServerlessDeployment
+from pinecone.core.openapi.db_control.model.pod_deployment import PodDeployment
+from pinecone.core.openapi.db_control.model.byoc_deployment import ByocDeployment
+from pinecone.core.openapi.db_control.model.index_model_status import IndexModelStatus
+from pinecone.db_control.models import IndexModel
+
+
+def make_index_model(
+    name: str = "test-index",
+    dimension: int | None = None,
+    metric: str | None = None,
+    spec: dict[str, Any] | None = None,
+    deployment: ServerlessDeployment | PodDeployment | ByocDeployment | None = None,
+    schema: Schema | None = None,
+    host: str = "https://test-index-1234.svc.pinecone.io",
+    status: IndexModelStatus | None = None,
+    **kwargs: Any,
+) -> IndexModel:
+    """Create an IndexModel for testing, accepting both old and new API parameters.
+
+    This helper function allows tests to use either the old-style parameters
+    (spec, dimension, metric) or new-style parameters (deployment, schema) for
+    backward compatibility.
+
+    Args:
+        name: The index name
+        dimension: Vector dimension (old-style, converted to schema)
+        metric: Distance metric (old-style, converted to schema)
+        spec: Index spec dict (old-style, converted to deployment)
+        deployment: Deployment configuration (new-style)
+        schema: Schema configuration (new-style)
+        host: Index host URL
+        status: Index status
+        **kwargs: Additional fields to pass to IndexModel
+
+    Returns:
+        IndexModel wrapper instance
+
+    Examples:
+        Old-style API::
+
+            index = make_index_model(
+                name="my-index",
+                dimension=1536,
+                metric="cosine",
+                spec={"serverless": {"cloud": "aws", "region": "us-east-1"}}
+            )
+
+        New-style API::
+
+            index = make_index_model(
+                name="my-index",
+                schema=Schema(fields={"_values": SchemaFields(
+                    type="dense_vector",
+                    dimension=1536,
+                    metric="cosine"
+                )}),
+                deployment=ServerlessDeployment(
+                    deployment_type="serverless",
+                    cloud="aws",
+                    region="us-east-1"
+                )
+            )
+    """
+    # Convert old-style spec to deployment if needed
+    if deployment is None and spec:
+        deployment = _spec_to_deployment(spec)
+    elif deployment is None:
+        # Default serverless deployment
+        deployment = ServerlessDeployment(
+            deployment_type="serverless", cloud="aws", region="us-east-1"
+        )
+
+    # Convert dimension/metric to schema if needed
+    if schema is None and (dimension or metric):
+        schema = _create_schema_from_dimension_metric(dimension, metric)
+    elif schema is None:
+        # Default empty schema (for FTS-only indexes)
+        schema = Schema(fields={})
+
+    # Create status if needed
+    if status is None:
+        status = IndexModelStatus(ready=True, state="Ready")
+
+    # Create the OpenAPI model
+    openapi_model = OpenAPIIndexModel(
+        name=name, schema=schema, deployment=deployment, host=host, status=status, **kwargs
+    )
+
+    # Wrap in the IndexModel wrapper
+    return IndexModel(openapi_model)
+
+
+def _spec_to_deployment(
+    spec: dict[str, Any],
+) -> ServerlessDeployment | PodDeployment | ByocDeployment:
+    """Convert old-style spec dict to new-style Deployment.
+
+    Args:
+        spec: Spec dict with serverless, pod, or byoc key
+
+    Returns:
+        Appropriate Deployment instance
+    """
+    if "serverless" in spec:
+        serverless_spec = spec["serverless"]
+        return ServerlessDeployment(
+            deployment_type="serverless",
+            cloud=serverless_spec.get("cloud", "aws"),
+            region=serverless_spec.get("region", "us-east-1"),
+        )
+    elif "pod" in spec:
+        pod_spec = spec["pod"]
+        return PodDeployment(
+            deployment_type="pod",
+            environment=pod_spec.get("environment", "us-east-1-aws"),
+            pod_type=pod_spec.get("pod_type", "p1.x1"),
+            replicas=pod_spec.get("replicas", 1),
+            shards=pod_spec.get("shards", 1),
+            pods=pod_spec.get("pods", 1),
+        )
+    elif "byoc" in spec:
+        byoc_spec = spec["byoc"]
+        return ByocDeployment(
+            deployment_type="byoc", environment=byoc_spec.get("environment", "custom-env")
+        )
+    else:
+        raise ValueError("spec must contain 'serverless', 'pod', or 'byoc' key")
+
+
+def _create_schema_from_dimension_metric(dimension: int | None, metric: str | None) -> Schema:
+    """Create a Schema from dimension and metric.
+
+    Args:
+        dimension: Vector dimension
+        metric: Distance metric
+
+    Returns:
+        Schema with a dense_vector field
+    """
+    if dimension is None:
+        # No dimension means FTS-only or sparse vectors
+        # For simplicity, return empty schema
+        return Schema(fields={})
+
+    field_config: dict[str, Any] = {"type": "dense_vector", "dimension": dimension}
+
+    if metric:
+        field_config["metric"] = metric
+
+    return Schema(fields={"_values": SchemaFields(**field_config)})
diff --git a/tests/unit/db_control/test_index.py b/tests/unit/db_control/test_index.py
index e30541044..f3882a88a 100644
--- a/tests/unit/db_control/test_index.py
+++ b/tests/unit/db_control/test_index.py
@@ -36,14 +36,19 @@ def test_describe_index(self, mocker):
         {
             "name": "test-index",
             "description": "test-description",
-            "dimension": 1024,
-            "metric": "cosine",
-            "spec": {
-                "byoc": {
-                    "environment": "test-environment"
+            "schema": {
+                "fields": {
+                    "_values": {
+                        "type": "dense_vector",
+                        "dimension": 1024,
+                        "metric": "cosine"
+                    }
                 }
             },
-            "vector_type": "dense",
+            "deployment": {
+                "deployment_type": "byoc",
+                "environment": "test-environment"
+            },
             "status": {
                 "ready": true,
                 "state": "Ready"
@@ -60,9 +65,10 @@ def test_describe_index(self, mocker):
         desc = index_resource.describe(name="test-index")
         assert desc.name == "test-index"
         assert desc.description == "test-description"
+        # Test backward compatibility properties
         assert desc.dimension == 1024
         assert desc.metric == "cosine"
-        assert desc.spec["byoc"]["environment"] == "test-environment"
+        assert desc.spec.byoc.environment == "test-environment"
         assert desc.vector_type == "dense"
         assert desc.status.ready == True
         assert desc.deletion_protection == "disabled"
@@ -104,9 +110,16 @@ def test_create_with_spec_uses_legacy_path(self, mocker):
         """Test that create() with spec uses the legacy request factory method."""
         body = """{
             "name": "test-index",
-            "dimension": 1536,
-            "metric": "cosine",
-            "spec": {"serverless": {"cloud": "aws", "region": "us-east-1"}},
+            "schema": {
+                "fields": {
+                    "_values": {
+                        "type": "dense_vector",
+                        "dimension": 1536,
+                        "metric": "cosine"
+                    }
+                }
+            },
+            "deployment": {"deployment_type": "serverless", "cloud": "aws", "region": "us-east-1"},
             "status": {"ready": true, "state": "Ready"},
             "host": "test.pinecone.io"
         }"""
diff --git a/tests/unit/db_control/test_index_request_factory.py b/tests/unit/db_control/test_index_request_factory.py
index 7a322cc21..72e912d2a 100644
--- a/tests/unit/db_control/test_index_request_factory.py
+++ b/tests/unit/db_control/test_index_request_factory.py
@@ -22,10 +22,13 @@ def test_create_index_request_with_spec_byoc(self):
             spec=ByocSpec(environment="test-byoc-spec-id"),
         )
         assert req.name == "test-index"
-        assert req.metric == "cosine"
-        assert req.dimension == 1024
-        assert req.spec.byoc.environment == "test-byoc-spec-id"
-        assert req.vector_type == "dense"
+        # New API: dimension/metric are in schema.fields
+        assert req.schema.fields["_values"].dimension == 1024
+        assert req.schema.fields["_values"].metric == "cosine"
+        assert req.schema.fields["_values"].type == "dense_vector"
+        # New API: spec is now deployment
+        assert req.deployment.deployment_type == "byoc"
+        assert req.deployment.environment == "test-byoc-spec-id"
         assert req.deletion_protection == "disabled"
 
     def test_create_index_request_with_spec_serverless(self):
@@ -36,11 +39,14 @@ def test_create_index_request_with_spec_serverless(self):
             spec=ServerlessSpec(cloud="aws", region="us-east-1"),
         )
         assert req.name == "test-index"
-        assert req.metric == "cosine"
-        assert req.dimension == 1024
-        assert req.spec.serverless.cloud == "aws"
-        assert req.spec.serverless.region == "us-east-1"
-        assert req.vector_type == "dense"
+        # New API: dimension/metric are in schema.fields
+        assert req.schema.fields["_values"].dimension == 1024
+        assert req.schema.fields["_values"].metric == "cosine"
+        assert req.schema.fields["_values"].type == "dense_vector"
+        # New API: spec is now deployment
+        assert req.deployment.deployment_type == "serverless"
+        assert req.deployment.cloud == "aws"
+        assert req.deployment.region == "us-east-1"
         assert req.deletion_protection == "disabled"
 
     def test_create_index_request_with_spec_serverless_dict(self):
@@ -51,11 +57,14 @@ def test_create_index_request_with_spec_serverless_dict(self):
             spec={"serverless": {"cloud": "aws", "region": "us-east-1"}},
         )
         assert req.name == "test-index"
-        assert req.metric == "cosine"
-        assert req.dimension == 1024
-        assert req.spec.serverless.cloud == "aws"
-        assert req.spec.serverless.region == "us-east-1"
-        assert req.vector_type == "dense"
+        # New API: dimension/metric are in schema.fields
+        assert req.schema.fields["_values"].dimension == 1024
+        assert req.schema.fields["_values"].metric == "cosine"
+        assert req.schema.fields["_values"].type == "dense_vector"
+        # New API: spec is now deployment
+        assert req.deployment.deployment_type == "serverless"
+        assert req.deployment.cloud == "aws"
+        assert req.deployment.region == "us-east-1"
         assert req.deletion_protection == "disabled"
 
     def test_create_index_request_with_spec_serverless_dict_enums(self):
@@ -67,11 +76,14 @@ def test_create_index_request_with_spec_serverless_dict_enums(self):
             spec={"serverless": {"cloud": CloudProvider.AWS, "region": AwsRegion.US_EAST_1}},
         )
         assert req.name == "test-index"
-        assert req.metric == "cosine"
-        assert req.dimension == 1024
-        assert req.spec.serverless.cloud == "aws"
-        assert req.spec.serverless.region == "us-east-1"
-        assert req.vector_type == "dense"
+        # New API: dimension/metric are in schema.fields
+        assert req.schema.fields["_values"].dimension == 1024
+        assert req.schema.fields["_values"].metric == "cosine"
+        assert req.schema.fields["_values"].type == "dense_vector"
+        # New API: spec is now deployment
+        assert req.deployment.deployment_type == "serverless"
+        assert req.deployment.cloud == "aws"
+        assert req.deployment.region == "us-east-1"
         assert req.deletion_protection == "disabled"
 
     def test_create_index_request_with_spec_byoc_dict(self):
@@ -82,10 +94,13 @@ def test_create_index_request_with_spec_byoc_dict(self):
             spec={"byoc": {"environment": "test-byoc-spec-id"}},
         )
         assert req.name == "test-index"
-        assert req.metric == "cosine"
-        assert req.dimension == 1024
-        assert req.spec.byoc.environment == "test-byoc-spec-id"
-        assert req.vector_type == "dense"
+        # New API: dimension/metric are in schema.fields
+        assert req.schema.fields["_values"].dimension == 1024
+        assert req.schema.fields["_values"].metric == "cosine"
+        assert req.schema.fields["_values"].type == "dense_vector"
+        # New API: spec is now deployment
+        assert req.deployment.deployment_type == "byoc"
+        assert req.deployment.environment == "test-byoc-spec-id"
         assert req.deletion_protection == "disabled"
 
     def test_create_index_request_with_spec_pod(self):
@@ -94,14 +109,17 @@ def test_create_index_request_with_spec_pod(self):
             name="test-index",
             metric="cosine",
             dimension=1024,
-            spec=PodSpec(environment="us-west1-gcp", pod_type="p1.x1"),
+            spec=PodSpec(environment="us-west1-gcp", pod_type="p1.x1", pods=1),
         )
         assert req.name == "test-index"
-        assert req.metric == "cosine"
-        assert req.dimension == 1024
-        assert req.spec.pod.environment == "us-west1-gcp"
-        assert req.spec.pod.pod_type == "p1.x1"
-        assert req.vector_type == "dense"
+        # New API: dimension/metric are in schema.fields
+        assert req.schema.fields["_values"].dimension == 1024
+        assert req.schema.fields["_values"].metric == "cosine"
+        assert req.schema.fields["_values"].type == "dense_vector"
+        # New API: spec is now deployment
+        assert req.deployment.deployment_type == "pod"
+        assert req.deployment.environment == "us-west1-gcp"
+        assert req.deployment.pod_type == "p1.x1"
         assert req.deletion_protection == "disabled"
 
     def test_create_index_request_with_spec_pod_all_fields(self):
@@ -121,16 +139,19 @@ def test_create_index_request_with_spec_pod_all_fields(self):
             ),
         )
         assert req.name == "test-index"
-        assert req.metric == "cosine"
-        assert req.dimension == 1024
-        assert req.spec.pod.environment == "us-west1-gcp"
-        assert req.spec.pod.pod_type == "p1.x1"
-        assert req.spec.pod.pods == 2
-        assert req.spec.pod.replicas == 1
-        assert req.spec.pod.shards == 1
-        assert req.spec.pod.metadata_config.indexed == ["field1", "field2"]
-        assert req.spec.pod.source_collection == "my-collection"
-        assert req.vector_type == "dense"
+        # New API: dimension/metric are in schema.fields
+        assert req.schema.fields["_values"].dimension == 1024
+        assert req.schema.fields["_values"].metric == "cosine"
+        assert req.schema.fields["_values"].type == "dense_vector"
+        # New API: spec is now deployment
+        assert req.deployment.deployment_type == "pod"
+        assert req.deployment.environment == "us-west1-gcp"
+        assert req.deployment.pod_type == "p1.x1"
+        assert req.deployment.pods == 2
+        assert req.deployment.replicas == 1
+        assert req.deployment.shards == 1
+        assert req.deployment.metadata_config.indexed == ["field1", "field2"]
+        # Note: source_collection is no longer on deployment in new API
         assert req.deletion_protection == "disabled"
 
     def test_create_index_request_with_spec_pod_dict(self):
@@ -139,14 +160,17 @@ def test_create_index_request_with_spec_pod_dict(self):
             name="test-index",
             metric="cosine",
             dimension=1024,
-            spec={"pod": {"environment": "us-west1-gcp", "pod_type": "p1.x1"}},
+            spec={"pod": {"environment": "us-west1-gcp", "pod_type": "p1.x1", "pods": 1}},
         )
         assert req.name == "test-index"
-        assert req.metric == "cosine"
-        assert req.dimension == 1024
-        assert req.spec.pod.environment == "us-west1-gcp"
-        assert req.spec.pod.pod_type == "p1.x1"
-        assert req.vector_type == "dense"
+        # New API: dimension/metric are in schema.fields
+        assert req.schema.fields["_values"].dimension == 1024
+        assert req.schema.fields["_values"].metric == "cosine"
+        assert req.schema.fields["_values"].type == "dense_vector"
+        # New API: spec is now deployment
+        assert req.deployment.deployment_type == "pod"
+        assert req.deployment.environment == "us-west1-gcp"
+        assert req.deployment.pod_type == "p1.x1"
         assert req.deletion_protection == "disabled"
 
     def test_create_index_request_with_spec_pod_dict_enums(self):
@@ -156,15 +180,22 @@ def test_create_index_request_with_spec_pod_dict_enums(self):
             metric="cosine",
             dimension=1024,
             spec={
-                "pod": {"environment": PodIndexEnvironment.US_WEST1_GCP, "pod_type": PodType.P1_X1}
+                "pod": {
+                    "environment": PodIndexEnvironment.US_WEST1_GCP,
+                    "pod_type": PodType.P1_X1,
+                    "pods": 1,
+                }
             },
         )
         assert req.name == "test-index"
-        assert req.metric == "cosine"
-        assert req.dimension == 1024
-        assert req.spec.pod.environment == "us-west1-gcp"
-        assert req.spec.pod.pod_type == "p1.x1"
-        assert req.vector_type == "dense"
+        # New API: dimension/metric are in schema.fields
+        assert req.schema.fields["_values"].dimension == 1024
+        assert req.schema.fields["_values"].metric == "cosine"
+        assert req.schema.fields["_values"].type == "dense_vector"
+        # New API: spec is now deployment
+        assert req.deployment.deployment_type == "pod"
+        assert req.deployment.environment == "us-west1-gcp"
+        assert req.deployment.pod_type == "p1.x1"
         assert req.deletion_protection == "disabled"
 
     def test_create_index_request_with_spec_pod_with_metadata_config(self):
@@ -176,16 +207,20 @@ def test_create_index_request_with_spec_pod_with_metadata_config(self):
             spec=PodSpec(
                 environment="us-west1-gcp",
                 pod_type="p1.x1",
+                pods=1,
                 metadata_config={"indexed": ["genre", "year"]},
             ),
         )
         assert req.name == "test-index"
-        assert req.metric == "cosine"
-        assert req.dimension == 1024
-        assert req.spec.pod.environment == "us-west1-gcp"
-        assert req.spec.pod.pod_type == "p1.x1"
-        assert req.spec.pod.metadata_config.indexed == ["genre", "year"]
-        assert req.vector_type == "dense"
+        # New API: dimension/metric are in schema.fields
+        assert req.schema.fields["_values"].dimension == 1024
+        assert req.schema.fields["_values"].metric == "cosine"
+        assert req.schema.fields["_values"].type == "dense_vector"
+        # New API: spec is now deployment
+        assert req.deployment.deployment_type == "pod"
+        assert req.deployment.environment == "us-west1-gcp"
+        assert req.deployment.pod_type == "p1.x1"
+        assert req.deployment.metadata_config.indexed == ["genre", "year"]
         assert req.deletion_protection == "disabled"
 
     def test_parse_read_capacity_ondemand(self):
@@ -214,10 +249,11 @@ def test_parse_read_capacity_dedicated_with_manual(self):
             )
         )
         assert result.mode == "Dedicated"
-        assert result.dedicated.node_type == "t1"
-        assert result.dedicated.scaling == "Manual"
-        assert result.dedicated.manual.shards == 2
-        assert result.dedicated.manual.replicas == 3
+        # New API: flattened structure
+        assert result.node_type == "t1"
+        assert result.scaling.strategy == "Manual"
+        assert result.scaling.shards == 2
+        assert result.scaling.replicas == 3
 
     def test_parse_read_capacity_dedicated_missing_manual(self):
         """Test that missing manual configuration raises ValueError when scaling is Manual."""
@@ -355,7 +391,7 @@ def test_translate_pod_spec_to_deployment_and_schema(self):
 
     def test_translate_pod_spec_with_defaults(self):
         """Test translating PodSpec with default values."""
-        spec = PodSpec(environment="us-east-1-aws")
+        spec = PodSpec(environment="us-east-1-aws", pods=1)
         deployment, schema = PineconeDBControlRequestFactory._translate_legacy_request(
             spec=spec, dimension=768, metric="cosine", vector_type="dense"
         )
@@ -366,8 +402,8 @@ def test_translate_pod_spec_with_defaults(self):
             "pod_type": "p1.x1",  # Default
             "replicas": 1,  # Default
             "shards": 1,  # Default
+            "pods": 1,  # Required, default is 1
         }
-        assert "pods" not in deployment  # Should not be included if None
         assert schema == {
             "fields": {"_values": {"type": "dense_vector", "dimension": 768, "metric": "cosine"}}
         }
@@ -512,25 +548,25 @@ def test_translate_dict_spec_with_enum_values(self):
         assert deployment["cloud"] == "aws"  # Enum converted to string
         assert deployment["region"] == "us-east-1"  # Enum converted to string
 
-    def test_translate_pod_spec_with_zero_replicas(self):
-        """Test that zero replicas/shards are preserved (not converted to 1)."""
-        spec = PodSpec(environment="us-east-1-aws", replicas=0, shards=0)
+    def test_translate_pod_spec_with_explicit_replicas_shards(self):
+        """Test that explicit replicas/shards values are used."""
+        spec = PodSpec(environment="us-east-1-aws", replicas=3, shards=2, pods=1)
         deployment, schema = PineconeDBControlRequestFactory._translate_legacy_request(
             spec=spec, dimension=1536, metric="cosine", vector_type="dense"
         )
 
-        assert deployment["replicas"] == 0  # Zero preserved
-        assert deployment["shards"] == 0  # Zero preserved
+        assert deployment["replicas"] == 3
+        assert deployment["shards"] == 2
 
-    def test_translate_dict_spec_with_zero_replicas(self):
-        """Test that zero replicas/shards in dict specs are preserved."""
-        spec = {"pod": {"environment": "us-east-1-aws", "replicas": 0, "shards": 0}}
+    def test_translate_dict_spec_with_explicit_replicas_shards(self):
+        """Test that explicit replicas/shards in dict specs are used."""
+        spec = {"pod": {"environment": "us-east-1-aws", "replicas": 3, "shards": 2, "pods": 1}}
         deployment, schema = PineconeDBControlRequestFactory._translate_legacy_request(
             spec=spec, dimension=1536, metric="cosine", vector_type="dense"
         )
 
-        assert deployment["replicas"] == 0  # Zero preserved
-        assert deployment["shards"] == 0  # Zero preserved
+        assert deployment["replicas"] == 3
+        assert deployment["shards"] == 2
 
     def test_translate_invalid_vector_type_raises_error(self):
         """Test that invalid vector_type raises ValueError instead of silently failing."""
diff --git a/tests/unit/models/test_index_list.py b/tests/unit/models/test_index_list.py
index bdd9b2842..1e7c6d49b 100644
--- a/tests/unit/models/test_index_list.py
+++ b/tests/unit/models/test_index_list.py
@@ -1,17 +1,14 @@
 import pytest
 from pinecone import IndexList
-from pinecone.core.openapi.db_control.models import (
-    IndexList as OpenApiIndexList,
-    IndexModel as OpenApiIndexModel,
-    IndexModelStatus,
-)
+from pinecone.core.openapi.db_control.models import IndexList as OpenApiIndexList, IndexModelStatus
+from tests.fixtures import make_index_model
 
 
 @pytest.fixture
 def index_list_response():
     return OpenApiIndexList(
         indexes=[
-            OpenApiIndexModel(
+            make_index_model(
                 name="test-index-1",
                 dimension=2,
                 metric="cosine",
@@ -27,8 +24,8 @@ def index_list_response():
                         "shards": 1,
                     }
                 },
-            ),
-            OpenApiIndexModel(
+            ).index,
+            make_index_model(
                 name="test-index-2",
                 dimension=3,
                 metric="cosine",
@@ -44,7 +41,7 @@ def index_list_response():
                         "shards": 1,
                     }
                 },
-            ),
+            ).index,
         ],
         _check_type=False,
     )
@@ -68,8 +65,9 @@ def test_index_list_getitem(self, index_list_response):
         iil = IndexList(index_list_response)
         input = index_list_response
         assert input.indexes[0].name == iil[0].name
-        assert input.indexes[0].dimension == iil[0].dimension
-        assert input.indexes[0].metric == iil[0].metric
+        # Test wrapped IndexModel compatibility properties
+        assert iil[0].dimension == 2
+        assert iil[0].metric == "cosine"
         assert input.indexes[0].host == iil[0].host
         assert input.indexes[0].deletion_protection == iil[0].deletion_protection
         assert iil[0].deletion_protection == "enabled"
diff --git a/tests/unit/openapi_support/test_api_client.py b/tests/unit/openapi_support/test_api_client.py
index 35abc74bc..c074f1dd6 100644
--- a/tests/unit/openapi_support/test_api_client.py
+++ b/tests/unit/openapi_support/test_api_client.py
@@ -1,4 +1,4 @@
-from pinecone.core.openapi.db_control.models import IndexModel, IndexModelStatus
+from pinecone.core.openapi.db_control.models import IndexModelStatus
 from pinecone.core.openapi.db_data.models import VectorValues
 from pinecone.openapi_support.serializer import Serializer
 from pinecone.openapi_support.api_client_utils import parameters_to_tuples
@@ -55,43 +55,40 @@ def test_sanitization_for_serialization_serializes_io(self):
         assert Serializer.sanitize_for_serialization(io.BytesIO(b"test")) == b"test"
 
     def test_sanitize_for_serialization_serializes_model_normal(self):
-        m = IndexModel(
+        from tests.fixtures import make_index_model
+
+        m = make_index_model(
             name="myindex",
             dimension=10,
             metric="cosine",
             host="localhost",
-            spec={},
             status=IndexModelStatus(ready=True, state="Ready"),
-            vector_type="dense",
-        )
-        assert Serializer.sanitize_for_serialization(m) == {
-            "name": "myindex",
-            "dimension": 10,
-            "metric": "cosine",
-            "host": "localhost",
-            "spec": {},
-            "status": {"ready": True, "state": "Ready"},
-            "vector_type": "dense",
-        }
-
-        m2 = IndexModel(
+        ).index  # Get the underlying OpenAPI model
+
+        serialized = Serializer.sanitize_for_serialization(m)
+        # New API format: uses schema and deployment
+        assert serialized["name"] == "myindex"
+        assert serialized["schema"]["fields"]["_values"]["dimension"] == 10
+        assert serialized["schema"]["fields"]["_values"]["metric"] == "cosine"
+        assert serialized["host"] == "localhost"
+        assert serialized["deployment"]["deployment_type"] == "serverless"
+        assert serialized["status"] == {"ready": True, "state": "Ready"}
+
+        m2 = make_index_model(
             name="myindex2",
             metric="cosine",
             host="localhost",
-            spec={},
             status=IndexModelStatus(ready=True, state="Ready"),
-            vector_type="sparse",
             deletion_protection="enabled",
-        )
-        assert Serializer.sanitize_for_serialization(m2) == {
-            "name": "myindex2",
-            "metric": "cosine",
-            "host": "localhost",
-            "spec": {},
-            "status": {"ready": True, "state": "Ready"},
-            "vector_type": "sparse",
-            "deletion_protection": "enabled",
-        }
+        ).index
+
+        serialized2 = Serializer.sanitize_for_serialization(m2)
+        assert serialized2["name"] == "myindex2"
+        assert serialized2["schema"]["fields"] == {}  # No dimension means empty schema
+        assert serialized2["host"] == "localhost"
+        assert serialized2["deployment"]["deployment_type"] == "serverless"
+        assert serialized2["status"] == {"ready": True, "state": "Ready"}
+        assert serialized2["deletion_protection"] == "enabled"
 
     def test_sanitize_for_serialization_serializes_model_simple(self):
         # ModelSimple is used to model named values which are not objects
diff --git a/tests/unit/test_control.py b/tests/unit/test_control.py
index a48544d87..e14b3f416 100644
--- a/tests/unit/test_control.py
+++ b/tests/unit/test_control.py
@@ -11,8 +11,9 @@
     PodIndexEnvironment,
     PodType,
 )
-from pinecone.core.openapi.db_control.models import IndexList, IndexModel, IndexModelStatus
+from pinecone.core.openapi.db_control.models import IndexList, IndexModelStatus
 from pinecone.utils import PluginAware
+from tests.fixtures import make_index_model
 
 
 import time
@@ -20,7 +21,7 @@
 
 def description_with_status(status: bool):
     state = "Ready" if status else "Initializing"
-    return IndexModel(
+    return make_index_model(
         name="foo",
         status=IndexModelStatus(ready=status, state=state),
         dimension=10,
@@ -33,38 +34,39 @@ def description_with_status(status: bool):
 
 @pytest.fixture
 def index_list_response():
+    from pinecone.core.openapi.db_control.model.index_model_status import (
+        IndexModelStatus as OpenAPIStatus,
+    )
+
     return IndexList(
         indexes=[
-            IndexModel(
+            make_index_model(
                 name="index1",
                 dimension=10,
                 metric="euclidean",
                 host="asdf.pinecone.io",
-                status={"ready": True},
+                status=OpenAPIStatus(ready=True, state="Ready"),
                 spec={},
                 deletion_protection="enabled",
-                _check_type=False,
-            ),
-            IndexModel(
+            ).index,
+            make_index_model(
                 name="index2",
                 dimension=10,
                 metric="euclidean",
                 host="asdf.pinecone.io",
-                status={"ready": True},
+                status=OpenAPIStatus(ready=True, state="Ready"),
                 spec={},
                 deletion_protection="enabled",
-                _check_type=False,
-            ),
-            IndexModel(
+            ).index,
+            make_index_model(
                 name="index3",
                 dimension=10,
                 metric="euclidean",
                 host="asdf.pinecone.io",
-                status={"ready": True},
+                status=OpenAPIStatus(ready=True, state="Ready"),
                 spec={},
                 deletion_protection="disabled",
-                _check_type=False,
-            ),
+            ).index,
         ]
     )
 
@@ -169,13 +171,22 @@ def test_create_index_with_timeout(
             ServerlessSpec(cloud=CloudProvider.GCP, region=GcpRegion.US_CENTRAL1),
             {"serverless": {"cloud": "aws", "region": "us-west1"}},
             {"serverless": {"cloud": "aws", "region": "us-west1", "uknown_key": "value"}},
-            PodSpec(environment="us-west1-gcp", pod_type="p1.x1"),
-            PodSpec(environment=PodIndexEnvironment.US_WEST1_GCP, pod_type=PodType.P2_X2),
-            PodSpec(environment=PodIndexEnvironment.US_WEST1_GCP, pod_type="s1.x4"),
-            PodSpec(environment=PodIndexEnvironment.US_EAST1_AWS, pod_type="unknown-pod-type"),
+            PodSpec(environment="us-west1-gcp", pod_type="p1.x1", pods=1),
+            PodSpec(environment=PodIndexEnvironment.US_WEST1_GCP, pod_type=PodType.P2_X2, pods=1),
+            PodSpec(environment=PodIndexEnvironment.US_WEST1_GCP, pod_type="s1.x4", pods=1),
+            PodSpec(
+                environment=PodIndexEnvironment.US_EAST1_AWS, pod_type="unknown-pod-type", pods=1
+            ),
             PodSpec(environment="us-west1-gcp", pod_type="p1.x1", pods=2, replicas=1, shards=1),
-            {"pod": {"environment": "us-west1-gcp", "pod_type": "p1.x1"}},
-            {"pod": {"environment": "us-west1-gcp", "pod_type": "p1.x1", "unknown_key": "value"}},
+            {"pod": {"environment": "us-west1-gcp", "pod_type": "p1.x1", "pods": 1}},
+            {
+                "pod": {
+                    "environment": "us-west1-gcp",
+                    "pod_type": "p1.x1",
+                    "pods": 1,
+                    "unknown_key": "value",
+                }
+            },
             {
                 "pod": {
                     "environment": "us-west1-gcp",