"""Configuration for the Pinecone SDK."""
from __future__ import annotations
import functools
import logging
import os
import re
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, ClassVar, Literal
from pinecone._internal.validation import validate_timeout
from pinecone.errors.exceptions import PineconeValueError
if TYPE_CHECKING:
from collections.abc import Callable
logger = logging.getLogger(__name__)
GrpcScheme = Literal["http", "https"]
GRPC_SCHEMES: tuple[str, ...] = ("http", "https")
_VISIBLE = r"[\x21-\x7e]"
_HEADER_VALUE = re.compile(rf"{_VISIBLE}(?:[\t ]*{_VISIBLE})*")
def check_api_key_is_header_safe(api_key: str) -> None:
"""Raise if *api_key* cannot be sent as an HTTP header value.
An empty key is left to the caller's own "no API key" error. The message never
includes the key. Called by each client after it resolves its key, not by
:class:`PineconeConfig`, so a clients-only check cannot reject an unrelated
``PINECONE_API_KEY`` for the Admin client, which never sends it.
"""
if api_key and _HEADER_VALUE.fullmatch(api_key) is None:
raise PineconeValueError(
"The API key (api_key or PINECONE_API_KEY) is not a valid HTTP header value: it is "
"blank or whitespace-only, has leading or "
"trailing whitespace (such as a newline read from a file), or contains a "
"control or non-ASCII character. Strip the key, for example api_key=key.strip()."
)
def resolve_grpc_scheme(scheme: str | None) -> GrpcScheme | None:
"""Resolve the gRPC endpoint scheme from an explicit value or the environment.
Args:
scheme: The caller's explicit choice, or ``None`` to fall back to the
``PINECONE_GRPC_SCHEME`` environment variable.
Returns:
``"http"``, ``"https"``, or ``None`` when neither source names one, which
leaves the choice to the caller's own default. A resolved ``"http"``
scheme warns once per process when the gRPC endpoint is built, unless
the host is loopback or RFC 1918 private.
Raises:
PineconeValueError: If either source names a scheme other than ``http``
or ``https``.
"""
resolved = scheme if scheme is not None else os.environ.get("PINECONE_GRPC_SCHEME", "").strip()
if not resolved:
return None
if resolved == "http":
return "http"
if resolved == "https":
return "https"
raise PineconeValueError(
f"Invalid gRPC scheme {resolved!r}. Must be one of: {', '.join(GRPC_SCHEMES)}."
)
def normalize_host(host: str | None) -> str:
"""Normalize a host string by ensuring it has an https:// prefix.
- If host is None or empty, return "".
- If host doesn't start with http:// or https://, prepend https://.
- Existing http:// or https:// prefixes are preserved as-is.
- Stacked-scheme patterns (e.g. "https://https://foo.io") are repaired by
stripping outer schemes and keeping the innermost one.
Idempotent: normalizing an already-normalized host returns it unchanged.
"""
if not host:
return ""
# Strip outer schemes until at most one remains, so any depth of stacking
# collapses to a single scheme and the result is idempotent.
while True:
stripped = False
for outer in ("https://", "http://"):
if host.startswith(outer):
remainder = host[len(outer) :]
if remainder.startswith(("http://", "https://")):
host = remainder
stripped = True
break
if not stripped:
break
if not host.startswith(("http://", "https://")):
return f"https://{host}"
return host
def normalize_source_tag(tag: str | None) -> str:
"""Normalize a source tag string.
- Lowercase the input.
- Strip characters not in [a-z0-9_ :].
- Replace spaces with underscores.
"""
if not tag:
return ""
import re
lowered = tag.lower()
cleaned = re.sub(r"[^a-z0-9_ :]", "", lowered)
return cleaned.replace(" ", "_")
def _parse_additional_headers_env() -> dict[str, str]:
"""Parse PINECONE_ADDITIONAL_HEADERS env var as JSON."""
raw = os.environ.get("PINECONE_ADDITIONAL_HEADERS", "")
if not raw:
return {}
return dict(_parse_additional_headers(raw))
@functools.lru_cache(maxsize=8)
def _parse_additional_headers(raw: str) -> dict[str, str]:
# Cached so a malformed value warns once per process rather than on every
# config built from the env. The value carries header secrets, so it is
# never logged.
import json
try:
parsed = json.loads(raw)
except Exception as exc:
logger.warning(
"Failed to parse PINECONE_ADDITIONAL_HEADERS env var as JSON (%s), ignoring it",
type(exc).__name__,
)
return {}
if isinstance(parsed, dict):
return {str(k): str(v) for k, v in parsed.items()}
return {}
[docs]
@dataclass(frozen=True)
class RetryConfig:
"""Configuration for HTTP retry behavior.
Args:
max_retries: Number of retries after the initial attempt. Defaults to 3 (4 total attempts).
backoff_factor: Minimum delay floor in seconds between retries. The
decorrelated-jitter algorithm samples from
``uniform(backoff_factor, prev_delay * 3)`` capped at ``max_wait``.
Defaults to 0.25.
max_wait: Maximum backoff delay in seconds. Defaults to 60.0.
retryable_status_codes: HTTP status codes that trigger a retry. Defaults to
``{408, 429, 500, 502, 503, 504}``.
on_throttle: Internal SDK callback invoked with the request URL host on every
retryable response (including ones that will be retried). Used by the SDK
to wire adaptive concurrency limiters; not intended for user configuration.
"""
max_retries: int = 3
backoff_factor: float = 0.25
max_wait: float = 60.0
retryable_status_codes: frozenset[int] = field(
default_factory=lambda: frozenset({408, 429, 500, 502, 503, 504})
)
on_throttle: Callable[[str], None] | None = None
@dataclass(frozen=True)
class PineconeConfig:
"""SDK configuration with environment variable fallbacks.
Args:
api_key: Pinecone API key. Falls back to PINECONE_API_KEY env var.
host: API host URL. Falls back to PINECONE_CONTROLLER_HOST env var.
timeout: Request timeout in seconds. Defaults to 30.
additional_headers: Extra headers to include in every request.
source_tag: Source tag for User-Agent string.
proxy_url: HTTP proxy URL.
ssl_ca_certs: Path to CA certificate bundle.
ssl_verify: Whether to verify SSL certificates.
grpc_scheme: URL scheme used to dial the gRPC data plane, ``"http"`` or
``"https"``. Falls back to the PINECONE_GRPC_SCHEME env var, then to
``None``, which lets the gRPC client keep its own default. Dialling
``"http"`` against a host outside loopback and the RFC 1918 private
ranges warns once per process, since the API key and every payload
then cross a public network unencrypted.
"""
api_key: str = ""
host: str = ""
timeout: float = 30.0
additional_headers: dict[str, str] = field(default_factory=dict)
source_tag: str = ""
proxy_url: str = ""
proxy_headers: dict[str, str] = field(default_factory=dict)
ssl_ca_certs: str | None = None
ssl_verify: bool = True
grpc_scheme: GrpcScheme | None = None
connection_pool_maxsize: int = 0
retry_config: RetryConfig = field(default_factory=RetryConfig)
_SENSITIVE_HEADER_KEYS: ClassVar[frozenset[str]] = frozenset(
{"authorization", "api-key", "proxy-authorization"}
)
def _redact_headers(self, headers: dict[str, str]) -> dict[str, str]:
return {
k: "***" if k.lower() in self._SENSITIVE_HEADER_KEYS else v for k, v in headers.items()
}
def __repr__(self) -> str:
masked = f"...{self.api_key[-4:]}" if len(self.api_key) >= 4 else "***"
return (
f"PineconeConfig("
f"api_key='{masked}', "
f"host='{self.host}', "
f"timeout={self.timeout}, "
f"additional_headers={self._redact_headers(self.additional_headers)!r}, "
f"source_tag='{self.source_tag}', "
f"proxy_url='{self.proxy_url}', "
f"proxy_headers={self._redact_headers(self.proxy_headers)!r}, "
f"ssl_ca_certs={self.ssl_ca_certs!r}, "
f"ssl_verify={self.ssl_verify}, "
f"grpc_scheme={self.grpc_scheme!r}, "
f"connection_pool_maxsize={self.connection_pool_maxsize}"
f")"
)
def __post_init__(self) -> None:
validate_timeout("timeout", self.timeout)
if not self.api_key:
env_key = os.environ.get("PINECONE_API_KEY", "")
object.__setattr__(self, "api_key", env_key)
if not self.host:
env_host = os.environ.get("PINECONE_CONTROLLER_HOST", "")
object.__setattr__(self, "host", normalize_host(env_host))
else:
object.__setattr__(self, "host", normalize_host(self.host))
if not self.additional_headers:
env_headers = _parse_additional_headers_env()
if env_headers:
object.__setattr__(self, "additional_headers", env_headers)
if self.source_tag:
object.__setattr__(self, "source_tag", normalize_source_tag(self.source_tag))
object.__setattr__(self, "grpc_scheme", resolve_grpc_scheme(self.grpc_scheme))