pinecone.control.index_host_store 

 1from typing import Dict
 2from pinecone.config import Config
 3from pinecone.core.openapi.db_control.api.manage_indexes_api import (
 4    ManageIndexesApi as IndexOperationsApi,
 5)
 6from pinecone.openapi_support.exceptions import PineconeException
 7from pinecone.utils import normalize_host
 8
 9
10class SingletonMeta(type):
11    _instances: Dict[str, str] = {}
12
13    def __call__(cls, *args, **kwargs):
14        if cls not in cls._instances:
15            instance = super().__call__(*args, **kwargs)
16            cls._instances[cls] = instance
17        return cls._instances[cls]
18
19
20class IndexHostStore(metaclass=SingletonMeta):
21    _indexHosts: Dict[str, str]
22
23    def __init__(self) -> None:
24        self._indexHosts = {}
25
26    def _key(self, config: Config, index_name: str) -> str:
27        return ":".join([config.api_key, index_name])
28
29    def delete_host(self, config: Config, index_name: str):
30        key = self._key(config, index_name)
31        if key in self._indexHosts:
32            del self._indexHosts[key]
33
34    def key_exists(self, key: str) -> bool:
35        return key in self._indexHosts
36
37    def set_host(self, config: Config, index_name: str, host: str):
38        if host:
39            key = self._key(config, index_name)
40            self._indexHosts[key] = normalize_host(host)
41
42    def get_host(self, api: IndexOperationsApi, config: Config, index_name: str) -> str:
43        key = self._key(config, index_name)
44        if self.key_exists(key):
45            return self._indexHosts[key]
46        else:
47            description = api.describe_index(index_name)
48            self.set_host(config, index_name, description.host)
49            if not self.key_exists(key):
50                raise PineconeException(
51                    f"Could not get host for index: {index_name}. Call describe_index('{index_name}') to check the current status."
52                )
53            return self._indexHosts[key]
class SingletonMeta(builtins.type): 
11class SingletonMeta(type):
12    _instances: Dict[str, str] = {}
13
14    def __call__(cls, *args, **kwargs):
15        if cls not in cls._instances:
16            instance = super().__call__(*args, **kwargs)
17            cls._instances[cls] = instance
18        return cls._instances[cls]

type(object) -> the object's type type(name, bases, dict, **kwds) -> a new type

Inherited Members
builtins.type
type
mro
class IndexHostStore: 
21class IndexHostStore(metaclass=SingletonMeta):
22    _indexHosts: Dict[str, str]
23
24    def __init__(self) -> None:
25        self._indexHosts = {}
26
27    def _key(self, config: Config, index_name: str) -> str:
28        return ":".join([config.api_key, index_name])
29
30    def delete_host(self, config: Config, index_name: str):
31        key = self._key(config, index_name)
32        if key in self._indexHosts:
33            del self._indexHosts[key]
34
35    def key_exists(self, key: str) -> bool:
36        return key in self._indexHosts
37
38    def set_host(self, config: Config, index_name: str, host: str):
39        if host:
40            key = self._key(config, index_name)
41            self._indexHosts[key] = normalize_host(host)
42
43    def get_host(self, api: IndexOperationsApi, config: Config, index_name: str) -> str:
44        key = self._key(config, index_name)
45        if self.key_exists(key):
46            return self._indexHosts[key]
47        else:
48            description = api.describe_index(index_name)
49            self.set_host(config, index_name, description.host)
50            if not self.key_exists(key):
51                raise PineconeException(
52                    f"Could not get host for index: {index_name}. Call describe_index('{index_name}') to check the current status."
53                )
54            return self._indexHosts[key]
def delete_host(self, config: pinecone.config.config.Config, index_name: str): 
30    def delete_host(self, config: Config, index_name: str):
31        key = self._key(config, index_name)
32        if key in self._indexHosts:
33            del self._indexHosts[key]
def key_exists(self, key: str) -> bool: 
35    def key_exists(self, key: str) -> bool:
36        return key in self._indexHosts
def set_host( self, config: pinecone.config.config.Config, index_name: str, host: str): 
38    def set_host(self, config: Config, index_name: str, host: str):
39        if host:
40            key = self._key(config, index_name)
41            self._indexHosts[key] = normalize_host(host)
def get_host( self, api: pinecone.core.openapi.db_control.api.manage_indexes_api.ManageIndexesApi, config: pinecone.config.config.Config, index_name: str) -> str: 
43    def get_host(self, api: IndexOperationsApi, config: Config, index_name: str) -> str:
44        key = self._key(config, index_name)
45        if self.key_exists(key):
46            return self._indexHosts[key]
47        else:
48            description = api.describe_index(index_name)
49            self.set_host(config, index_name, description.host)
50            if not self.key_exists(key):
51                raise PineconeException(
52                    f"Could not get host for index: {index_name}. Call describe_index('{index_name}') to check the current status."
53                )
54            return self._indexHosts[key]