Phase 2A: Response Adapter Consolidation

pinecone/adapters/__init__.py

@@ -0,0 +1,28 @@
+"""Adapter layer for converting between OpenAPI models and SDK types.
+
+This module provides a centralized adapter layer that isolates SDK code from
+the details of generated OpenAPI models. This enables:
+
+- Single source of truth for response transformations
+- Easier testing (adapters can be tested in isolation)
+- Version flexibility (adapters can handle different API versions)
+- Clear contracts between generated and SDK code
+
+Usage:
+    >>> from pinecone.adapters import adapt_query_response, adapt_upsert_response
+    >>> sdk_response = adapt_query_response(openapi_response)
+"""
+
+from pinecone.adapters.response_adapters import (
+    adapt_fetch_response,
+    adapt_query_response,
+    adapt_upsert_response,
+    UpsertResponseTransformer,
+)
+
+__all__ = [
+    "adapt_fetch_response",
+    "adapt_query_response",
+    "adapt_upsert_response",
+    "UpsertResponseTransformer",
+]

pinecone/adapters/response_adapters.py

@@ -0,0 +1,151 @@
+"""Adapter functions for converting OpenAPI responses to SDK dataclasses.
+
+This module provides centralized functions for transforming OpenAPI response
+objects into SDK-specific dataclasses. These adapters isolate the SDK from
+changes in the OpenAPI model structure.
+
+The adapter functions replace duplicated parsing logic that was previously
+scattered across multiple modules.
+"""
+
+from __future__ import annotations
+
+from multiprocessing.pool import ApplyResult
+from typing import TYPE_CHECKING, Any
+
+from pinecone.adapters.utils import extract_response_metadata
+
+if TYPE_CHECKING:
+    from pinecone.db_data.dataclasses.fetch_response import FetchResponse
+    from pinecone.db_data.dataclasses.query_response import QueryResponse
+    from pinecone.db_data.dataclasses.upsert_response import UpsertResponse
+
+
+def adapt_query_response(openapi_response: Any) -> QueryResponse:
+    """Adapt an OpenAPI QueryResponse to the SDK QueryResponse dataclass.
+
+    This function extracts fields from the OpenAPI response object and
+    constructs an SDK-native QueryResponse dataclass. It handles:
+    - Extracting matches and namespace
+    - Optional usage information
+    - Response metadata (headers)
+    - Cleaning up deprecated 'results' field
+
+    Args:
+        openapi_response: An OpenAPI QueryResponse object from the generated code.
+
+    Returns:
+        A QueryResponse dataclass instance.
+
+    Example:
+        >>> from pinecone.adapters import adapt_query_response
+        >>> sdk_response = adapt_query_response(openapi_response)
+        >>> print(sdk_response.matches)
+    """
+    # Import at runtime to avoid circular imports
+    from pinecone.db_data.dataclasses.query_response import QueryResponse as QR
+
+    response_info = extract_response_metadata(openapi_response)
+
+    # Remove deprecated 'results' field if present
+    if hasattr(openapi_response, "_data_store"):
+        openapi_response._data_store.pop("results", None)
+
+    return QR(
+        matches=openapi_response.matches,
+        namespace=openapi_response.namespace or "",
+        usage=openapi_response.usage
+        if hasattr(openapi_response, "usage") and openapi_response.usage
+        else None,
+        _response_info=response_info,
+    )
+
+
+def adapt_upsert_response(openapi_response: Any) -> UpsertResponse:
+    """Adapt an OpenAPI UpsertResponse to the SDK UpsertResponse dataclass.
+
+    Args:
+        openapi_response: An OpenAPI UpsertResponse object from the generated code.
+
+    Returns:
+        An UpsertResponse dataclass instance.
+
+    Example:
+        >>> from pinecone.adapters import adapt_upsert_response
+        >>> sdk_response = adapt_upsert_response(openapi_response)
+        >>> print(sdk_response.upserted_count)
+    """
+    # Import at runtime to avoid circular imports
+    from pinecone.db_data.dataclasses.upsert_response import UpsertResponse as UR
+
+    response_info = extract_response_metadata(openapi_response)
+
+    return UR(upserted_count=openapi_response.upserted_count, _response_info=response_info)
+
+
+def adapt_fetch_response(openapi_response: Any) -> FetchResponse:
+    """Adapt an OpenAPI FetchResponse to the SDK FetchResponse dataclass.
+
+    This function extracts fields from the OpenAPI response object and
+    constructs an SDK-native FetchResponse dataclass. It handles:
+    - Converting vectors dict to SDK Vector objects
+    - Optional usage information
+    - Response metadata (headers)
+
+    Args:
+        openapi_response: An OpenAPI FetchResponse object from the generated code.
+
+    Returns:
+        A FetchResponse dataclass instance.
+
+    Example:
+        >>> from pinecone.adapters import adapt_fetch_response
+        >>> sdk_response = adapt_fetch_response(openapi_response)
+        >>> print(sdk_response.vectors)
+    """
+    # Import at runtime to avoid circular imports
+    from pinecone.db_data.dataclasses.fetch_response import FetchResponse as FR
+    from pinecone.db_data.dataclasses.vector import Vector
+
+    response_info = extract_response_metadata(openapi_response)
+
+    return FR(
+        namespace=openapi_response.namespace,
+        vectors={k: Vector.from_dict(v) for k, v in openapi_response.vectors.items()},
+        usage=openapi_response.usage,
+        _response_info=response_info,
+    )
+
+
+class UpsertResponseTransformer:
+    """Transformer for converting ApplyResult[OpenAPIUpsertResponse] to UpsertResponse.
+
+    This wrapper transforms the OpenAPI response to our dataclass when .get() is called,
+    while delegating other methods to the underlying ApplyResult.
+
+    Example:
+        >>> transformer = UpsertResponseTransformer(async_result)
+        >>> response = transformer.get()  # Returns UpsertResponse
+    """
+
+    _apply_result: ApplyResult
+    """ :meta private: """
+
+    def __init__(self, apply_result: ApplyResult) -> None:
+        self._apply_result = apply_result
+
+    def get(self, timeout: float | None = None) -> UpsertResponse:
+        """Get the transformed UpsertResponse.
+
+        Args:
+            timeout: Optional timeout in seconds for the underlying result.
+
+        Returns:
+            The SDK UpsertResponse dataclass.
+        """
+        openapi_response = self._apply_result.get(timeout)
+        return adapt_upsert_response(openapi_response)
+
+    def __getattr__(self, name: str) -> Any:
+        # Delegate other methods to the underlying ApplyResult
+        return getattr(self._apply_result, name)

pinecone/adapters/utils.py

@@ -0,0 +1,31 @@
+"""Shared utilities for adapter implementations.
+
+This module provides helper functions used across different adapters
+for common operations like extracting response metadata.
+"""
+
+from typing import Any
+
+from pinecone.utils.response_info import ResponseInfo, extract_response_info
+
+
+def extract_response_metadata(response: Any) -> ResponseInfo:
+    """Extract response metadata from an OpenAPI response object.
+
+    Extracts the _response_info attribute from an OpenAPI response if present,
+    otherwise returns an empty ResponseInfo with empty headers.
+
+    Args:
+        response: An OpenAPI response object that may have _response_info.
+
+    Returns:
+        ResponseInfo with extracted headers, or empty headers if not present.
+    """
+    response_info = None
+    if hasattr(response, "_response_info"):
+        response_info = response._response_info
+
+    if response_info is None:
+        response_info = extract_response_info({})
+
+    return response_info

pinecone/db_data/index.py

@@ -52,6 +52,11 @@
 )
 from .query_results_aggregator import QueryResultsAggregator, QueryNamespacesResults
 from pinecone.openapi_support import OPENAPI_ENDPOINT_PARAMS
+from pinecone.adapters.response_adapters import (
+    adapt_query_response,
+    adapt_fetch_response,
+    UpsertResponseTransformer,
+)
 
 from multiprocessing.pool import ApplyResult
 from multiprocessing import cpu_count
@@ -76,58 +81,12 @@
 
 
 def parse_query_response(response: OpenAPIQueryResponse) -> QueryResponse:
-    """:meta private:"""
-    # Convert OpenAPI QueryResponse to dataclass QueryResponse
-    from pinecone.utils.response_info import extract_response_info
-
-    response_info = None
-    if hasattr(response, "_response_info"):
-        response_info = response._response_info
-
-    if response_info is None:
-        response_info = extract_response_info({})
-
-    # Remove deprecated 'results' field if present
-    if hasattr(response, "_data_store"):
-        response._data_store.pop("results", None)
-
-    return QueryResponse(
-        matches=response.matches,
-        namespace=response.namespace or "",
-        usage=response.usage if hasattr(response, "usage") and response.usage else None,
-        _response_info=response_info,
-    )
-
-
-class UpsertResponseTransformer:
-    """Transformer for converting ApplyResult[OpenAPIUpsertResponse] to UpsertResponse.
+    """:meta private:
 
-    This wrapper transforms the OpenAPI response to our dataclass when .get() is called,
-    while delegating other methods to the underlying ApplyResult.
+    Deprecated: Use adapt_query_response from pinecone.adapters instead.
+    This function is kept for backward compatibility.
     """
-
-    _apply_result: ApplyResult
-    """ :meta private: """
-
-    def __init__(self, apply_result: ApplyResult) -> None:
-        self._apply_result = apply_result
-
-    def get(self, timeout: float | None = None) -> UpsertResponse:
-        openapi_response = self._apply_result.get(timeout)
-        from pinecone.utils.response_info import extract_response_info
-
-        response_info = None
-        if hasattr(openapi_response, "_response_info"):
-            response_info = openapi_response._response_info
-        if response_info is None:
-            response_info = extract_response_info({})
-        return UpsertResponse(
-            upserted_count=openapi_response.upserted_count, _response_info=response_info
-        )
-
-    def __getattr__(self, name: str) -> Any:
-        # Delegate other methods to the underlying ApplyResult
-        return getattr(self._apply_result, name)
+    return adapt_query_response(response)
 
 
 class Index(PluginAware):
@@ -894,22 +853,7 @@ def fetch(self, ids: list[str], namespace: str | None = None, **kwargs) -> Fetch
         """
         args_dict = parse_non_empty_args([("namespace", namespace)])
         result = self._vector_api.fetch_vectors(ids=ids, **args_dict, **kwargs)
-        # Copy response info from OpenAPI response if present
-        from pinecone.utils.response_info import extract_response_info
-
-        response_info = None
-        if hasattr(result, "_response_info"):
-            response_info = result._response_info
-        if response_info is None:
-            response_info = extract_response_info({})
-
-        fetch_response = FetchResponse(
-            namespace=result.namespace,
-            vectors={k: Vector.from_dict(v) for k, v in result.vectors.items()},
-            usage=result.usage,
-            _response_info=response_info,
-        )
-        return fetch_response
+        return adapt_fetch_response(result)
 
     @validate_and_convert_errors
     def fetch_by_metadata(

pinecone/db_data/index_asyncio.py

@@ -58,6 +58,7 @@
 )
 
 from pinecone.openapi_support import OPENAPI_ENDPOINT_PARAMS
+from pinecone.adapters.response_adapters import adapt_query_response, adapt_fetch_response
 from .index import IndexRequestFactory
 
 from .vector_factory import VectorFactory
@@ -91,27 +92,12 @@
 
 
 def parse_query_response(response: OpenAPIQueryResponse) -> QueryResponse:
-    """:meta private:"""
-    # Convert OpenAPI QueryResponse to dataclass QueryResponse
-    from pinecone.utils.response_info import extract_response_info
-
-    response_info = None
-    if hasattr(response, "_response_info"):
-        response_info = response._response_info
-
-    if response_info is None:
-        response_info = extract_response_info({})
-
-    # Remove deprecated 'results' field if present
-    if hasattr(response, "_data_store"):
-        response._data_store.pop("results", None)
-
-    return QueryResponse(
-        matches=response.matches,
-        namespace=response.namespace or "",
-        usage=response.usage if hasattr(response, "usage") and response.usage else None,
-        _response_info=response_info,
-    )
+    """:meta private:
+
+    Deprecated: Use adapt_query_response from pinecone.adapters instead.
+    This function is kept for backward compatibility.
+    """
+    return adapt_query_response(response)
 
 
 class _IndexAsyncio:
@@ -647,22 +633,7 @@ async def main():
         """
         args_dict = parse_non_empty_args([("namespace", namespace)])
         result = await self._vector_api.fetch_vectors(ids=ids, **args_dict, **kwargs)
-        # Copy response info from OpenAPI response if present
-        from pinecone.utils.response_info import extract_response_info
-
-        response_info = None
-        if hasattr(result, "_response_info"):
-            response_info = result._response_info
-        if response_info is None:
-            response_info = extract_response_info({})
-
-        fetch_response = FetchResponse(
-            namespace=result.namespace,
-            vectors={k: Vector.from_dict(v) for k, v in result.vectors.items()},
-            usage=result.usage,
-            _response_info=response_info,
-        )
-        return fetch_response
+        return adapt_fetch_response(result)
 
     @validate_and_convert_errors
     async def fetch_by_metadata(