pinecone.control.index_host_store 

 1from typing import Dict
 2from pinecone.config import Config
 3from pinecone.core.openapi.control.api.manage_indexes_api import (
 4    ManageIndexesApi as IndexOperationsApi,
 5)
 6from pinecone.core.openapi.shared.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    def __init__(self):
22        self._indexHosts = {}
23
24    def _key(self, config: Config, index_name: str) -> str:
25        return ":".join([config.api_key, index_name])
26
27    def delete_host(self, config: Config, index_name: str):
28        key = self._key(config, index_name)
29        if key in self._indexHosts:
30            del self._indexHosts[key]
31
32    def key_exists(self, key: str) -> bool:
33        return key in self._indexHosts
34
35    def set_host(self, config: Config, index_name: str, host: str):
36        if host:
37            key = self._key(config, index_name)
38            self._indexHosts[key] = normalize_host(host)
39
40    def get_host(self, api: IndexOperationsApi, config: Config, index_name: str) -> str:
41        key = self._key(config, index_name)
42        if self.key_exists(key):
43            return self._indexHosts[key]
44        else:
45            description = api.describe_index(index_name)
46            self.set_host(config, index_name, description.host)
47            if not self.key_exists(key):
48                raise PineconeException(
49                    f"Could not get host for index: {index_name}. Call describe_index('{index_name}') to check the current status."
50                )
51            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    def __init__(self):
23        self._indexHosts = {}
24
25    def _key(self, config: Config, index_name: str) -> str:
26        return ":".join([config.api_key, index_name])
27
28    def delete_host(self, config: Config, index_name: str):
29        key = self._key(config, index_name)
30        if key in self._indexHosts:
31            del self._indexHosts[key]
32
33    def key_exists(self, key: str) -> bool:
34        return key in self._indexHosts
35
36    def set_host(self, config: Config, index_name: str, host: str):
37        if host:
38            key = self._key(config, index_name)
39            self._indexHosts[key] = normalize_host(host)
40
41    def get_host(self, api: IndexOperationsApi, config: Config, index_name: str) -> str:
42        key = self._key(config, index_name)
43        if self.key_exists(key):
44            return self._indexHosts[key]
45        else:
46            description = api.describe_index(index_name)
47            self.set_host(config, index_name, description.host)
48            if not self.key_exists(key):
49                raise PineconeException(
50                    f"Could not get host for index: {index_name}. Call describe_index('{index_name}') to check the current status."
51                )
52            return self._indexHosts[key]
def delete_host(self, config: pinecone.config.config.Config, index_name: str): 
28    def delete_host(self, config: Config, index_name: str):
29        key = self._key(config, index_name)
30        if key in self._indexHosts:
31            del self._indexHosts[key]
def key_exists(self, key: str) -> bool: 
33    def key_exists(self, key: str) -> bool:
34        return key in self._indexHosts
def set_host( self, config: pinecone.config.config.Config, index_name: str, host: str): 
36    def set_host(self, config: Config, index_name: str, host: str):
37        if host:
38            key = self._key(config, index_name)
39            self._indexHosts[key] = normalize_host(host)
def get_host( self, api: pinecone.core.openapi.control.api.manage_indexes_api.ManageIndexesApi, config: pinecone.config.config.Config, index_name: str) -> str: 
41    def get_host(self, api: IndexOperationsApi, config: Config, index_name: str) -> str:
42        key = self._key(config, index_name)
43        if self.key_exists(key):
44            return self._indexHosts[key]
45        else:
46            description = api.describe_index(index_name)
47            self.set_host(config, index_name, description.host)
48            if not self.key_exists(key):
49                raise PineconeException(
50                    f"Could not get host for index: {index_name}. Call describe_index('{index_name}') to check the current status."
51                )
52            return self._indexHosts[key]