basic upsert

upsert working?

temp routing name

fetch

list

type name

inline request body for upsert

upsert request body

_id magic

.

Simple HTTP impl

update route name

updates

Author: ssmith-pc

pinecone/pinecone.py

@@ -16,6 +16,7 @@
 if TYPE_CHECKING:
     from pinecone.config import Config, OpenApiConfiguration
     from pinecone.db_data import _Index as Index, _IndexAsyncio as IndexAsyncio
+    from pinecone.repository.data import _Repository as Repository
     from pinecone.db_control.index_host_store import IndexHostStore
     from pinecone.core.openapi.db_control.api.manage_indexes_api import ManageIndexesApi
     from pinecone.db_control.types import CreateIndexForModelEmbedTypedDict, ConfigureIndexEmbed
@@ -518,6 +519,35 @@ def IndexAsyncio(self, host: str, **kwargs) -> "IndexAsyncio":
             **kwargs,
         )
 
+    def Repository(self, name: str = "", host: str = "", **kwargs) -> "Repository":
+        from pinecone.repository.data import _Repository
+
+        if name == "" and host == "":
+            raise ValueError("Either name or host must be specified")
+
+        pt = kwargs.pop("pool_threads", None) or self._pool_threads
+        api_key = self._config.api_key
+        openapi_config = self._openapi_config
+
+        if host != "":
+            check_realistic_host(host)
+
+            # Use host url if it is provided
+            repository_host = normalize_host(host)
+        else:
+            # TODO, get host url from describe_kb using the index name
+            # index_host = self.db.index._get_host(name)
+            raise ValueError("host lookup not yet supported, specify host parameter")
+
+        return _Repository(
+            host=repository_host,
+            api_key=api_key,
+            pool_threads=pt,
+            openapi_config=openapi_config,
+            source_tag=self.config.source_tag,
+            **kwargs,
+        )
+
 
 def check_realistic_host(host: str) -> None:
     """:meta private:

pinecone/repository/data/__init__.py

@@ -0,0 +1,13 @@
+from .repository import Repository
+
+
+_Repository = Repository  # alias for backwards compatibility
+
+
+__all__ = ["_Repository"]
+
+
+def __getattr__(name):
+    if name in locals():
+        return locals()[name]
+    raise AttributeError(f"module '{__name__}' has no attribute '{name}'")

pinecone/repository/data/repository.py

@@ -0,0 +1,205 @@
+import logging
+import json
+from typing import Optional, Dict, Any, Tuple
+from urllib.parse import urljoin
+
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+from multiprocessing import cpu_count
+from pinecone.core.openapi.repository_data import API_VERSION
+
+logger = logging.getLogger(__name__)
+
+
+def _ensure_https_host(host: str) -> str:
+    """
+    Normalizes the host value to include scheme and no trailing slash.
+    Accepts: "kb.example.com", "https://kb.example.com/", "http://..."
+    Returns: "https://kb.example.com"
+    """
+    host = (host or "").strip()
+    if not host:
+        raise ValueError("host must be provided (e.g., 'kb.your-company.com').")
+    if not host.startswith(("http://", "https://")):
+        host = "https://" + host
+    # strip single trailing slash
+    if host.endswith("/"):
+        host = host[:-1]
+    return host
+
+
+class HTTPError(Exception):
+    """Rich HTTP error including status code and server payload (if any)."""
+
+    def __init__(self, status_code: int, message: str, payload: Optional[dict] = None):
+        super().__init__(f"{status_code}: {message}")
+        self.status_code = status_code
+        self.payload = payload or {}
+
+
+class Repository:
+    """
+    A client for interacting with the Pinecone Knowledge Base Data Plane (Documents).
+    Uses `requests` directly, with retries and sane defaults.
+
+    Methods return plain `dict` responses parsed from JSON.
+    """
+
+    def __init__(
+        self,
+        api_key: str,
+        host: str,
+        pool_threads: Optional[int] = None,
+        additional_headers: Optional[Dict[str, str]] = None,
+        openapi_config=None,  # kept for backward compat; unused
+        echo: bool = False,
+        **kwargs,
+    ):
+        self._api_key = api_key
+        self._base_url = _ensure_https_host(host)
+        self._echo = echo  # store the flag
+
+        # Connection pool sizing
+        self._pool_threads = 5 * cpu_count() if pool_threads is None else pool_threads
+        pool_maxsize = kwargs.get("connection_pool_maxsize", self._pool_threads)
+
+        # Timeouts (connect, read). Allow overrides via kwargs
+        # e.g., timeout=(3.05, 30)
+        self._timeout: Tuple[float, float] = kwargs.get("timeout", (5.0, 60.0))
+
+        # Retries: conservative defaults; override via kwargs["retries"]
+        retries = kwargs.get(
+            "retries",
+            Retry(
+                total=5,
+                backoff_factor=0.5,
+                status_forcelist=(429, 500, 502, 503, 504),
+                allowed_methods=frozenset(["GET", "POST", "DELETE"]),
+                raise_on_status=False,
+            ),
+        )
+
+        self._session = requests.Session()
+        adapter = HTTPAdapter(
+            pool_connections=self._pool_threads, pool_maxsize=pool_maxsize, max_retries=retries
+        )
+        self._session.mount("https://", adapter)
+        self._session.mount("http://", adapter)
+
+        self._default_headers = {
+            "Api-Key": self._api_key,
+            "Accept": "application/json",
+            "x-pinecone-api-version": API_VERSION,
+            # Content-Type set per request when needed
+        }
+        if additional_headers:
+            self._default_headers.update(additional_headers)
+
+    # -----------------------
+    # Internal request helper
+    # -----------------------
+    def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        json_body: Optional[dict] = None,
+        headers: Optional[dict] = None,
+        params: Optional[dict] = None,
+        echo: Optional[bool] = None,
+    ) -> dict:
+        url = urljoin(self._base_url + "/", path.lstrip("/"))
+        hdrs = dict(self._default_headers)
+        if headers:
+            hdrs.update(headers)
+        if json_body is not None:
+            hdrs.setdefault("Content-Type", "application/json")
+
+        logger.debug("HTTP %s %s params=%s json=%s", method, url, params, json_body)
+
+        # decide whether to echo this call
+        do_echo = self._echo if echo is None else echo
+        if do_echo:
+            print("----- HTTP Request -----")
+            print(f"{method} {url}")
+            if params:
+                print("Params:", params)
+
+            safe_headers = dict(hdrs)
+            for k, v in hdrs.items():
+                print(f"checking........... {k}: {v}")
+                if k.lower() == "api-key":
+                    masked = (v[:5] + "...") if isinstance(v, str) and len(v) > 5 else "..."
+                    safe_headers[k] = masked
+                else:
+                    safe_headers[k] = v
+
+            print("Headers:", safe_headers)
+            if json_body is not None:
+                print("Body:", json.dumps(json_body, indent=2))
+            print("------------------------")
+
+        resp = self._session.request(
+            method=method,
+            url=url,
+            headers=hdrs,
+            params=params,
+            json=json_body,
+            timeout=self._timeout,
+        )
+
+        # Try to parse JSON payload (even on errors) for better messages
+        payload: Optional[dict]
+        try:
+            payload = resp.json() if resp.content else None
+        except json.JSONDecodeError:
+            payload = None
+
+        if not (200 <= resp.status_code < 300):
+            msg = payload.get("message") if isinstance(payload, dict) else resp.text
+            raise HTTPError(resp.status_code, msg or "HTTP request failed", payload)
+
+        if payload is None:
+            return {}
+        return payload
+
+    # -------------
+    # API methods
+    # -------------
+    def upsert(self, namespace: str, document: Dict[str, Any], **kwargs) -> dict:
+        """
+        POST /knowledge-bases//namespaces/{namespace}/documents/upsert
+        Returns UpsertDocumentResponse as dict.
+        """
+        if not isinstance(document, dict):
+            raise TypeError("document must be a dict (JSON-serializable).")
+
+        path = f"/knowledge-bases/{namespace}/documents/upsert"
+        return self._request("POST", path, json_body=document, **kwargs)
+
+    def fetch(self, namespace: str, document_id: str, **kwargs) -> dict:
+        """
+        GET /knowledge-bases/{namespace}/documents/{document_id}
+        Returns GetDocumentResponse as dict.
+        """
+        path = f"/knowledge-bases/{namespace}/documents/{document_id}"
+        return self._request("GET", path, **kwargs)
+
+    def list(self, namespace: str, **kwargs) -> dict:
+        """
+        GET /knowledge-bases/{namespace}/documents
+        Returns ListDocumentsResponse as dict.
+        """
+        path = f"/knowledge-bases/{namespace}/documents"
+        # Spec does not define query params, but keep hook if server adds (e.g., pagination).
+        params = kwargs.get("params")
+        return self._request("GET", path, params=params, **kwargs)
+
+    def delete(self, namespace: str, document_id: str, **kwargs) -> dict:
+        """
+        DELETE /knowledge-bases/{namespace}/documents/{document_id}
+        Returns DeleteDocumentResponse as dict.
+        """
+        path = f"/knowledge-bases/{namespace}/documents/{document_id}"
+        return self._request("DELETE", path, **kwargs)