pinecone.control.pinecone

View Source

  1import time
  2import logging
  3from typing import Optional, Dict, Union
  4from multiprocessing import cpu_count
  5
  6from .index_host_store import IndexHostStore
  7from .pinecone_interface import PineconeDBControlInterface
  8
  9from pinecone.config import PineconeConfig, ConfigBuilder
 10
 11from pinecone.core.openapi.db_control.api.manage_indexes_api import ManageIndexesApi
 12from pinecone.openapi_support.api_client import ApiClient
 13
 14
 15from pinecone.utils import normalize_host, setup_openapi_client, PluginAware
 16from pinecone.core.openapi.db_control import API_VERSION
 17from pinecone.models import (
 18    ServerlessSpec,
 19    PodSpec,
 20    IndexModel,
 21    IndexList,
 22    CollectionList,
 23    IndexEmbed,
 24)
 25from .langchain_import_warnings import _build_langchain_attribute_error_message
 26from pinecone.utils import docslinks
 27from pinecone.data import _Index, _Inference, _IndexAsyncio
 28
 29from pinecone.enums import (
 30    Metric,
 31    VectorType,
 32    DeletionProtection,
 33    PodType,
 34    CloudProvider,
 35    AwsRegion,
 36    GcpRegion,
 37    AzureRegion,
 38)
 39from .types import CreateIndexForModelEmbedTypedDict
 40from .request_factory import PineconeDBControlRequestFactory
 41
 42logger = logging.getLogger(__name__)
 43""" @private """
 44
 45
 46class Pinecone(PineconeDBControlInterface, PluginAware):
 47    """
 48    A client for interacting with Pinecone's vector database.
 49
 50    This class implements methods for managing and interacting with Pinecone resources
 51    such as collections and indexes.
 52    """
 53
 54    def __init__(
 55        self,
 56        api_key: Optional[str] = None,
 57        host: Optional[str] = None,
 58        proxy_url: Optional[str] = None,
 59        proxy_headers: Optional[Dict[str, str]] = None,
 60        ssl_ca_certs: Optional[str] = None,
 61        ssl_verify: Optional[bool] = None,
 62        additional_headers: Optional[Dict[str, str]] = {},
 63        pool_threads: Optional[int] = None,
 64        **kwargs,
 65    ):
 66        for deprecated_kwarg in {"config", "openapi_config", "index_api"}:
 67            if deprecated_kwarg in kwargs:
 68                raise NotImplementedError(
 69                    f"Passing {deprecated_kwarg} is no longer supported. Please pass individual settings such as proxy_url, proxy_headers, ssl_ca_certs, and ssl_verify directly to the Pinecone constructor as keyword arguments. See the README at {docslinks['README']} for examples."
 70                )
 71
 72        self.config = PineconeConfig.build(
 73            api_key=api_key,
 74            host=host,
 75            additional_headers=additional_headers,
 76            proxy_url=proxy_url,
 77            proxy_headers=proxy_headers,
 78            ssl_ca_certs=ssl_ca_certs,
 79            ssl_verify=ssl_verify,
 80            **kwargs,
 81        )
 82        """ @private """
 83
 84        self.openapi_config = ConfigBuilder.build_openapi_config(self.config, **kwargs)
 85        """ @private """
 86
 87        if pool_threads is None:
 88            self.pool_threads = 5 * cpu_count()
 89            """ @private """
 90        else:
 91            self.pool_threads = pool_threads
 92            """ @private """
 93
 94        self._inference = None  # Lazy initialization
 95        """ @private """
 96
 97        self.index_api = setup_openapi_client(
 98            api_client_klass=ApiClient,
 99            api_klass=ManageIndexesApi,
100            config=self.config,
101            openapi_config=self.openapi_config,
102            pool_threads=pool_threads,
103            api_version=API_VERSION,
104        )
105        """ @private """
106
107        self.index_host_store = IndexHostStore()
108        """ @private """
109
110        self.load_plugins(
111            config=self.config, openapi_config=self.openapi_config, pool_threads=self.pool_threads
112        )
113
114    @property
115    def inference(self):
116        """
117        Inference is a namespace where an instance of the `pinecone.data.features.inference.inference.Inference` class is lazily created and cached.
118        """
119        if self._inference is None:
120            self._inference = _Inference(config=self.config, openapi_config=self.openapi_config)
121        return self._inference
122
123    def create_index(
124        self,
125        name: str,
126        spec: Union[Dict, ServerlessSpec, PodSpec],
127        dimension: Optional[int] = None,
128        metric: Optional[Union[Metric, str]] = Metric.COSINE,
129        timeout: Optional[int] = None,
130        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
131        vector_type: Optional[Union[VectorType, str]] = VectorType.DENSE,
132        tags: Optional[Dict[str, str]] = None,
133    ) -> IndexModel:
134        req = PineconeDBControlRequestFactory.create_index_request(
135            name=name,
136            spec=spec,
137            dimension=dimension,
138            metric=metric,
139            deletion_protection=deletion_protection,
140            vector_type=vector_type,
141            tags=tags,
142        )
143        resp = self.index_api.create_index(create_index_request=req)
144
145        if timeout == -1:
146            return IndexModel(resp)
147        return self.__poll_describe_index_until_ready(name, timeout)
148
149    def create_index_for_model(
150        self,
151        name: str,
152        cloud: Union[CloudProvider, str],
153        region: Union[AwsRegion, GcpRegion, AzureRegion, str],
154        embed: Union[IndexEmbed, CreateIndexForModelEmbedTypedDict],
155        tags: Optional[Dict[str, str]] = None,
156        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
157        timeout: Optional[int] = None,
158    ) -> IndexModel:
159        req = PineconeDBControlRequestFactory.create_index_for_model_request(
160            name=name,
161            cloud=cloud,
162            region=region,
163            embed=embed,
164            tags=tags,
165            deletion_protection=deletion_protection,
166        )
167        resp = self.index_api.create_index_for_model(req)
168
169        if timeout == -1:
170            return IndexModel(resp)
171        return self.__poll_describe_index_until_ready(name, timeout)
172
173    def __poll_describe_index_until_ready(self, name: str, timeout: Optional[int] = None):
174        description = None
175
176        def is_ready() -> bool:
177            nonlocal description
178            description = self.describe_index(name=name)
179            return description.status.ready
180
181        total_wait_time = 0
182        if timeout is None:
183            # Wait indefinitely
184            while not is_ready():
185                logger.debug(
186                    f"Waiting for index {name} to be ready. Total wait time {total_wait_time} seconds."
187                )
188                total_wait_time += 5
189                time.sleep(5)
190
191        else:
192            # Wait for a maximum of timeout seconds
193            while not is_ready():
194                if timeout < 0:
195                    logger.error(f"Index {name} is not ready. Timeout reached.")
196                    link = docslinks["API_DESCRIBE_INDEX"]
197                    timeout_msg = (
198                        f"Please call describe_index() to confirm index status. See docs at {link}"
199                    )
200                    raise TimeoutError(timeout_msg)
201
202                logger.debug(
203                    f"Waiting for index {name} to be ready. Total wait time: {total_wait_time}"
204                )
205                total_wait_time += 5
206                time.sleep(5)
207                timeout -= 5
208
209        return description
210
211    def delete_index(self, name: str, timeout: Optional[int] = None):
212        self.index_api.delete_index(name)
213        self.index_host_store.delete_host(self.config, name)
214
215        if timeout == -1:
216            return
217
218        if timeout is None:
219            while self.has_index(name):
220                time.sleep(5)
221        else:
222            while self.has_index(name) and timeout >= 0:
223                time.sleep(5)
224                timeout -= 5
225        if timeout and timeout < 0:
226            raise (
227                TimeoutError(
228                    "Please call the list_indexes API ({}) to confirm if index is deleted".format(
229                        "https://www.pinecone.io/docs/api/operation/list_indexes/"
230                    )
231                )
232            )
233
234    def list_indexes(self) -> IndexList:
235        response = self.index_api.list_indexes()
236        return IndexList(response)
237
238    def describe_index(self, name: str) -> IndexModel:
239        api_instance = self.index_api
240        description = api_instance.describe_index(name)
241        host = description.host
242        self.index_host_store.set_host(self.config, name, host)
243
244        return IndexModel(description)
245
246    def has_index(self, name: str) -> bool:
247        if name in self.list_indexes().names():
248            return True
249        else:
250            return False
251
252    def configure_index(
253        self,
254        name: str,
255        replicas: Optional[int] = None,
256        pod_type: Optional[Union[PodType, str]] = None,
257        deletion_protection: Optional[Union[DeletionProtection, str]] = None,
258        tags: Optional[Dict[str, str]] = None,
259    ):
260        api_instance = self.index_api
261        description = self.describe_index(name=name)
262
263        req = PineconeDBControlRequestFactory.configure_index_request(
264            description=description,
265            replicas=replicas,
266            pod_type=pod_type,
267            deletion_protection=deletion_protection,
268            tags=tags,
269        )
270        api_instance.configure_index(name, configure_index_request=req)
271
272    def create_collection(self, name: str, source: str) -> None:
273        req = PineconeDBControlRequestFactory.create_collection_request(name=name, source=source)
274        self.index_api.create_collection(create_collection_request=req)
275
276    def list_collections(self) -> CollectionList:
277        response = self.index_api.list_collections()
278        return CollectionList(response)
279
280    def delete_collection(self, name: str) -> None:
281        self.index_api.delete_collection(name)
282
283    def describe_collection(self, name: str):
284        return self.index_api.describe_collection(name).to_dict()
285
286    @staticmethod
287    def from_texts(*args, **kwargs):
288        """@private"""
289        raise AttributeError(_build_langchain_attribute_error_message("from_texts"))
290
291    @staticmethod
292    def from_documents(*args, **kwargs):
293        """@private"""
294        raise AttributeError(_build_langchain_attribute_error_message("from_documents"))
295
296    def Index(self, name: str = "", host: str = "", **kwargs):
297        if name == "" and host == "":
298            raise ValueError("Either name or host must be specified")
299
300        pt = kwargs.pop("pool_threads", None) or self.pool_threads
301        api_key = self.config.api_key
302        openapi_config = self.openapi_config
303
304        if host != "":
305            check_realistic_host(host)
306
307            # Use host url if it is provided
308            index_host = normalize_host(host)
309        else:
310            # Otherwise, get host url from describe_index using the index name
311            index_host = self.index_host_store.get_host(self.index_api, self.config, name)
312
313        return _Index(
314            host=index_host,
315            api_key=api_key,
316            pool_threads=pt,
317            openapi_config=openapi_config,
318            source_tag=self.config.source_tag,
319            **kwargs,
320        )
321
322    def IndexAsyncio(self, host: str, **kwargs):
323        api_key = self.config.api_key
324        openapi_config = self.openapi_config
325
326        if host is None or host == "":
327            raise ValueError("A host must be specified")
328
329        check_realistic_host(host)
330        index_host = normalize_host(host)
331
332        return _IndexAsyncio(
333            host=index_host,
334            api_key=api_key,
335            openapi_config=openapi_config,
336            source_tag=self.config.source_tag,
337            **kwargs,
338        )
339
340
341def check_realistic_host(host: str) -> None:
342    """@private
343
344    Checks whether a user-provided host string seems plausible.
345    Someone could erroneously pass an index name as the host by
346    mistake, and if they have done that we'd like to give them a
347    simple error message as feedback rather than attempting to
348    call the url and getting a more cryptic DNS resolution error.
349    """
350
351    if "." not in host and "localhost" not in host:
352        raise ValueError(
353            f"You passed '{host}' as the host but this does not appear to be valid. Call describe_index() to confirm the host of the index."
354        )

class Pinecone(pinecone.control.pinecone_interface.PineconeDBControlInterface, pinecone.utils.plugin_aware.PluginAware): View Source

 47class Pinecone(PineconeDBControlInterface, PluginAware):
 48    """
 49    A client for interacting with Pinecone's vector database.
 50
 51    This class implements methods for managing and interacting with Pinecone resources
 52    such as collections and indexes.
 53    """
 54
 55    def __init__(
 56        self,
 57        api_key: Optional[str] = None,
 58        host: Optional[str] = None,
 59        proxy_url: Optional[str] = None,
 60        proxy_headers: Optional[Dict[str, str]] = None,
 61        ssl_ca_certs: Optional[str] = None,
 62        ssl_verify: Optional[bool] = None,
 63        additional_headers: Optional[Dict[str, str]] = {},
 64        pool_threads: Optional[int] = None,
 65        **kwargs,
 66    ):
 67        for deprecated_kwarg in {"config", "openapi_config", "index_api"}:
 68            if deprecated_kwarg in kwargs:
 69                raise NotImplementedError(
 70                    f"Passing {deprecated_kwarg} is no longer supported. Please pass individual settings such as proxy_url, proxy_headers, ssl_ca_certs, and ssl_verify directly to the Pinecone constructor as keyword arguments. See the README at {docslinks['README']} for examples."
 71                )
 72
 73        self.config = PineconeConfig.build(
 74            api_key=api_key,
 75            host=host,
 76            additional_headers=additional_headers,
 77            proxy_url=proxy_url,
 78            proxy_headers=proxy_headers,
 79            ssl_ca_certs=ssl_ca_certs,
 80            ssl_verify=ssl_verify,
 81            **kwargs,
 82        )
 83        """ @private """
 84
 85        self.openapi_config = ConfigBuilder.build_openapi_config(self.config, **kwargs)
 86        """ @private """
 87
 88        if pool_threads is None:
 89            self.pool_threads = 5 * cpu_count()
 90            """ @private """
 91        else:
 92            self.pool_threads = pool_threads
 93            """ @private """
 94
 95        self._inference = None  # Lazy initialization
 96        """ @private """
 97
 98        self.index_api = setup_openapi_client(
 99            api_client_klass=ApiClient,
100            api_klass=ManageIndexesApi,
101            config=self.config,
102            openapi_config=self.openapi_config,
103            pool_threads=pool_threads,
104            api_version=API_VERSION,
105        )
106        """ @private """
107
108        self.index_host_store = IndexHostStore()
109        """ @private """
110
111        self.load_plugins(
112            config=self.config, openapi_config=self.openapi_config, pool_threads=self.pool_threads
113        )
114
115    @property
116    def inference(self):
117        """
118        Inference is a namespace where an instance of the `pinecone.data.features.inference.inference.Inference` class is lazily created and cached.
119        """
120        if self._inference is None:
121            self._inference = _Inference(config=self.config, openapi_config=self.openapi_config)
122        return self._inference
123
124    def create_index(
125        self,
126        name: str,
127        spec: Union[Dict, ServerlessSpec, PodSpec],
128        dimension: Optional[int] = None,
129        metric: Optional[Union[Metric, str]] = Metric.COSINE,
130        timeout: Optional[int] = None,
131        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
132        vector_type: Optional[Union[VectorType, str]] = VectorType.DENSE,
133        tags: Optional[Dict[str, str]] = None,
134    ) -> IndexModel:
135        req = PineconeDBControlRequestFactory.create_index_request(
136            name=name,
137            spec=spec,
138            dimension=dimension,
139            metric=metric,
140            deletion_protection=deletion_protection,
141            vector_type=vector_type,
142            tags=tags,
143        )
144        resp = self.index_api.create_index(create_index_request=req)
145
146        if timeout == -1:
147            return IndexModel(resp)
148        return self.__poll_describe_index_until_ready(name, timeout)
149
150    def create_index_for_model(
151        self,
152        name: str,
153        cloud: Union[CloudProvider, str],
154        region: Union[AwsRegion, GcpRegion, AzureRegion, str],
155        embed: Union[IndexEmbed, CreateIndexForModelEmbedTypedDict],
156        tags: Optional[Dict[str, str]] = None,
157        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
158        timeout: Optional[int] = None,
159    ) -> IndexModel:
160        req = PineconeDBControlRequestFactory.create_index_for_model_request(
161            name=name,
162            cloud=cloud,
163            region=region,
164            embed=embed,
165            tags=tags,
166            deletion_protection=deletion_protection,
167        )
168        resp = self.index_api.create_index_for_model(req)
169
170        if timeout == -1:
171            return IndexModel(resp)
172        return self.__poll_describe_index_until_ready(name, timeout)
173
174    def __poll_describe_index_until_ready(self, name: str, timeout: Optional[int] = None):
175        description = None
176
177        def is_ready() -> bool:
178            nonlocal description
179            description = self.describe_index(name=name)
180            return description.status.ready
181
182        total_wait_time = 0
183        if timeout is None:
184            # Wait indefinitely
185            while not is_ready():
186                logger.debug(
187                    f"Waiting for index {name} to be ready. Total wait time {total_wait_time} seconds."
188                )
189                total_wait_time += 5
190                time.sleep(5)
191
192        else:
193            # Wait for a maximum of timeout seconds
194            while not is_ready():
195                if timeout < 0:
196                    logger.error(f"Index {name} is not ready. Timeout reached.")
197                    link = docslinks["API_DESCRIBE_INDEX"]
198                    timeout_msg = (
199                        f"Please call describe_index() to confirm index status. See docs at {link}"
200                    )
201                    raise TimeoutError(timeout_msg)
202
203                logger.debug(
204                    f"Waiting for index {name} to be ready. Total wait time: {total_wait_time}"
205                )
206                total_wait_time += 5
207                time.sleep(5)
208                timeout -= 5
209
210        return description
211
212    def delete_index(self, name: str, timeout: Optional[int] = None):
213        self.index_api.delete_index(name)
214        self.index_host_store.delete_host(self.config, name)
215
216        if timeout == -1:
217            return
218
219        if timeout is None:
220            while self.has_index(name):
221                time.sleep(5)
222        else:
223            while self.has_index(name) and timeout >= 0:
224                time.sleep(5)
225                timeout -= 5
226        if timeout and timeout < 0:
227            raise (
228                TimeoutError(
229                    "Please call the list_indexes API ({}) to confirm if index is deleted".format(
230                        "https://www.pinecone.io/docs/api/operation/list_indexes/"
231                    )
232                )
233            )
234
235    def list_indexes(self) -> IndexList:
236        response = self.index_api.list_indexes()
237        return IndexList(response)
238
239    def describe_index(self, name: str) -> IndexModel:
240        api_instance = self.index_api
241        description = api_instance.describe_index(name)
242        host = description.host
243        self.index_host_store.set_host(self.config, name, host)
244
245        return IndexModel(description)
246
247    def has_index(self, name: str) -> bool:
248        if name in self.list_indexes().names():
249            return True
250        else:
251            return False
252
253    def configure_index(
254        self,
255        name: str,
256        replicas: Optional[int] = None,
257        pod_type: Optional[Union[PodType, str]] = None,
258        deletion_protection: Optional[Union[DeletionProtection, str]] = None,
259        tags: Optional[Dict[str, str]] = None,
260    ):
261        api_instance = self.index_api
262        description = self.describe_index(name=name)
263
264        req = PineconeDBControlRequestFactory.configure_index_request(
265            description=description,
266            replicas=replicas,
267            pod_type=pod_type,
268            deletion_protection=deletion_protection,
269            tags=tags,
270        )
271        api_instance.configure_index(name, configure_index_request=req)
272
273    def create_collection(self, name: str, source: str) -> None:
274        req = PineconeDBControlRequestFactory.create_collection_request(name=name, source=source)
275        self.index_api.create_collection(create_collection_request=req)
276
277    def list_collections(self) -> CollectionList:
278        response = self.index_api.list_collections()
279        return CollectionList(response)
280
281    def delete_collection(self, name: str) -> None:
282        self.index_api.delete_collection(name)
283
284    def describe_collection(self, name: str):
285        return self.index_api.describe_collection(name).to_dict()
286
287    @staticmethod
288    def from_texts(*args, **kwargs):
289        """@private"""
290        raise AttributeError(_build_langchain_attribute_error_message("from_texts"))
291
292    @staticmethod
293    def from_documents(*args, **kwargs):
294        """@private"""
295        raise AttributeError(_build_langchain_attribute_error_message("from_documents"))
296
297    def Index(self, name: str = "", host: str = "", **kwargs):
298        if name == "" and host == "":
299            raise ValueError("Either name or host must be specified")
300
301        pt = kwargs.pop("pool_threads", None) or self.pool_threads
302        api_key = self.config.api_key
303        openapi_config = self.openapi_config
304
305        if host != "":
306            check_realistic_host(host)
307
308            # Use host url if it is provided
309            index_host = normalize_host(host)
310        else:
311            # Otherwise, get host url from describe_index using the index name
312            index_host = self.index_host_store.get_host(self.index_api, self.config, name)
313
314        return _Index(
315            host=index_host,
316            api_key=api_key,
317            pool_threads=pt,
318            openapi_config=openapi_config,
319            source_tag=self.config.source_tag,
320            **kwargs,
321        )
322
323    def IndexAsyncio(self, host: str, **kwargs):
324        api_key = self.config.api_key
325        openapi_config = self.openapi_config
326
327        if host is None or host == "":
328            raise ValueError("A host must be specified")
329
330        check_realistic_host(host)
331        index_host = normalize_host(host)
332
333        return _IndexAsyncio(
334            host=index_host,
335            api_key=api_key,
336            openapi_config=openapi_config,
337            source_tag=self.config.source_tag,
338            **kwargs,
339        )

A client for interacting with Pinecone's vector database.

This class implements methods for managing and interacting with Pinecone resources such as collections and indexes.

Pinecone( api_key: Optional[str] = None, host: Optional[str] = None, proxy_url: Optional[str] = None, proxy_headers: Optional[Dict[str, str]] = None, ssl_ca_certs: Optional[str] = None, ssl_verify: Optional[bool] = None, additional_headers: Optional[Dict[str, str]] = {}, pool_threads: Optional[int] = None, **kwargs) View Source

 55    def __init__(
 56        self,
 57        api_key: Optional[str] = None,
 58        host: Optional[str] = None,
 59        proxy_url: Optional[str] = None,
 60        proxy_headers: Optional[Dict[str, str]] = None,
 61        ssl_ca_certs: Optional[str] = None,
 62        ssl_verify: Optional[bool] = None,
 63        additional_headers: Optional[Dict[str, str]] = {},
 64        pool_threads: Optional[int] = None,
 65        **kwargs,
 66    ):
 67        for deprecated_kwarg in {"config", "openapi_config", "index_api"}:
 68            if deprecated_kwarg in kwargs:
 69                raise NotImplementedError(
 70                    f"Passing {deprecated_kwarg} is no longer supported. Please pass individual settings such as proxy_url, proxy_headers, ssl_ca_certs, and ssl_verify directly to the Pinecone constructor as keyword arguments. See the README at {docslinks['README']} for examples."
 71                )
 72
 73        self.config = PineconeConfig.build(
 74            api_key=api_key,
 75            host=host,
 76            additional_headers=additional_headers,
 77            proxy_url=proxy_url,
 78            proxy_headers=proxy_headers,
 79            ssl_ca_certs=ssl_ca_certs,
 80            ssl_verify=ssl_verify,
 81            **kwargs,
 82        )
 83        """ @private """
 84
 85        self.openapi_config = ConfigBuilder.build_openapi_config(self.config, **kwargs)
 86        """ @private """
 87
 88        if pool_threads is None:
 89            self.pool_threads = 5 * cpu_count()
 90            """ @private """
 91        else:
 92            self.pool_threads = pool_threads
 93            """ @private """
 94
 95        self._inference = None  # Lazy initialization
 96        """ @private """
 97
 98        self.index_api = setup_openapi_client(
 99            api_client_klass=ApiClient,
100            api_klass=ManageIndexesApi,
101            config=self.config,
102            openapi_config=self.openapi_config,
103            pool_threads=pool_threads,
104            api_version=API_VERSION,
105        )
106        """ @private """
107
108        self.index_host_store = IndexHostStore()
109        """ @private """
110
111        self.load_plugins(
112            config=self.config, openapi_config=self.openapi_config, pool_threads=self.pool_threads
113        )

The Pinecone class is the main entry point for interacting with Pinecone via this Python SDK. Instances of the Pinecone class are used to create, delete, and manage your indexes and collections. The class also holds inference functionality (embed, rerank) under the inference namespace. Methods which create or modify index and collection resources result in network calls to https://api pinecone.control.pinecone.io.

When you are ready to perform data operations on an index, you will need to instantiate an index client. Though the functionality of the index client is defined in a different class, it is instantiated through the Index() method in order for configurations to be shared between the two objects.

Parameters

api_key: The API key to use for authentication. If not passed via kwarg, the API key will be read from the environment variable PINECONE_API_KEY.
host: The control plane host. If unspecified, the host api.pinecone.io will be used.
proxy_url: The URL of the proxy to use for the connection.
proxy_headers: Additional headers to pass to the proxy. Use this if your proxy setup requires authentication.
ssl_ca_certs: The path to the SSL CA certificate bundle to use for the connection. This path should point to a file in PEM format. When not passed, the SDK will use the certificate bundle returned from certifi.where().
ssl_verify: SSL verification is performed by default, but can be disabled using the boolean flag when testing with Pinecone Local or troubleshooting a proxy setup. You should never run with SSL verification disabled in production.
additional_headers: Additional headers to pass to the API. This is mainly to support internal testing at Pinecone. End users should not need to use this unless following specific instructions to do so.
pool_threads: The number of threads to use for the ThreadPool when using methods that support the async_req keyword argument. The default number of threads is 5 * the number of CPUs in your execution environment.

Configuration with environment variables

If you instantiate the Pinecone client with no arguments, it will attempt to read the API key from the environment variable PINECONE_API_KEY.

from pinecone import Pinecone

pc = Pinecone()

Configuration with keyword arguments

If you prefer being more explicit in your code, you can also pass the API key as a keyword argument. This is also where you will pass additional configuration options such as proxy settings if you wish to use those.

import os
from pinecone import Pinecone

pc = Pinecone(
    api_key=os.environ.get("PINECONE_API_KEY"),
    host="https://api-staging.pinecone.io"
)

Environment variables

The Pinecone client supports the following environment variables:

PINECONE_API_KEY: The API key to use for authentication. If not passed via kwarg, the API key will be read from the environment variable PINECONE_API_KEY.
PINECONE_DEBUG_CURL: When troubleshooting it can be very useful to run curl commands against the control plane API to see exactly what data is being sent and received without all the abstractions and transformations applied by the Python SDK. If you set this environment variable to true, the Pinecone client will use request parameters to print out an equivalent curl command that you can run yourself or share with Pinecone support. Be very careful with this option, as it will print out your API key which forms part of a required authentication header. The main use of is to help evaluate whether a problem you are experiencing is due to the API's behavior or the behavior of the SDK itself.

Proxy configuration

If your network setup requires you to interact with Pinecone via a proxy, you will need to pass additional configuration using optional keyword parameters. These optional parameters are forwarded to urllib3, which is the underlying library currently used by the Pinecone client to make HTTP requests. You may find it helpful to refer to the urllib3 documentation on working with proxies while troubleshooting these settings.

Here is a basic example:

from pinecone import Pinecone

pc = Pinecone(
    api_key='YOUR_API_KEY',
    proxy_url='https://your-proxy.com'
)

pc.list_indexes()

If your proxy requires authentication, you can pass those values in a header dictionary using the proxy_headers parameter.

from pinecone import Pinecone
import urllib3 import make_headers

pc = Pinecone(
    api_key='YOUR_API_KEY',
    proxy_url='https://your-proxy.com',
    proxy_headers=make_headers(proxy_basic_auth='username:password')
)

pc.list_indexes()

Using proxies with self-signed certificates

By default the Pinecone Python client will perform SSL certificate verification using the CA bundle maintained by Mozilla in the certifi package. If your proxy server is using a self-signed certificate, you will need to pass the path to the certificate in PEM format using the ssl_ca_certs parameter.

from pinecone import Pinecone
import urllib3 import make_headers

pc = Pinecone(
    api_key='YOUR_API_KEY',
    proxy_url='https://your-proxy.com',
    proxy_headers=make_headers(proxy_basic_auth='username:password'),
    ssl_ca_certs='path/to/cert-bundle.pem'
)

pc.list_indexes()

Disabling SSL verification

If you would like to disable SSL verification, you can pass the ssl_verify parameter with a value of False. We do not recommend going to production with SSL verification disabled.

from pinecone import Pinecone
import urllib3 import make_headers

pc = Pinecone(
    api_key='YOUR_API_KEY',
    proxy_url='https://your-proxy.com',
    proxy_headers=make_headers(proxy_basic_auth='username:password'),
    ssl_ca_certs='path/to/cert-bundle.pem',
    ssl_verify=False
)

pc.list_indexes()

inference View Source

115    @property
116    def inference(self):
117        """
118        Inference is a namespace where an instance of the `pinecone.data.features.inference.inference.Inference` class is lazily created and cached.
119        """
120        if self._inference is None:
121            self._inference = _Inference(config=self.config, openapi_config=self.openapi_config)
122        return self._inference

Inference is a namespace where an instance of the pinecone.data.features.inference.inference.Inference class is lazily created and cached.

def create_index( self, name: str, spec: Union[Dict, pinecone.models.serverless_spec.ServerlessSpec, pinecone.models.pod_spec.PodSpec], dimension: Optional[int] = None, metric: Union[pinecone.enums.metric.Metric, str, NoneType] = <Metric.COSINE: 'cosine'>, timeout: Optional[int] = None, deletion_protection: Union[pinecone.enums.deletion_protection.DeletionProtection, str, NoneType] = <DeletionProtection.DISABLED: 'disabled'>, vector_type: Union[pinecone.enums.vector_type.VectorType, str, NoneType] = <VectorType.DENSE: 'dense'>, tags: Optional[Dict[str, str]] = None) -> pinecone.models.index_model.IndexModel: View Source

124    def create_index(
125        self,
126        name: str,
127        spec: Union[Dict, ServerlessSpec, PodSpec],
128        dimension: Optional[int] = None,
129        metric: Optional[Union[Metric, str]] = Metric.COSINE,
130        timeout: Optional[int] = None,
131        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
132        vector_type: Optional[Union[VectorType, str]] = VectorType.DENSE,
133        tags: Optional[Dict[str, str]] = None,
134    ) -> IndexModel:
135        req = PineconeDBControlRequestFactory.create_index_request(
136            name=name,
137            spec=spec,
138            dimension=dimension,
139            metric=metric,
140            deletion_protection=deletion_protection,
141            vector_type=vector_type,
142            tags=tags,
143        )
144        resp = self.index_api.create_index(create_index_request=req)
145
146        if timeout == -1:
147            return IndexModel(resp)
148        return self.__poll_describe_index_until_ready(name, timeout)

Creates a Pinecone index.

Parameters

name: The name of the index to create. Must be unique within your project and cannot be changed once created. Allowed characters are lowercase letters, numbers, and hyphens and the name may not begin or end with hyphens. Maximum length is 45 characters.
metric: Type of similarity metric used in the vector index when querying, one of {"cosine", "dotproduct", "euclidean"}.
spec: A dictionary containing configurations describing how the index should be deployed. For serverless indexes, specify region and cloud. For pod indexes, specify replicas, shards, pods, pod_type, metadata_config, and source_collection. Alternatively, use the ServerlessSpec or PodSpec objects to specify these configurations.
dimension: If you are creating an index with vector_type="dense" (which is the default), you need to specify dimension to indicate the size of your vectors. This should match the dimension of the embeddings you will be inserting. For example, if you are using OpenAI's CLIP model, you should use dimension=1536. Dimension is a required field when creating an index with vector_type="dense" and should not be passed when vector_type="sparse".
timeout: Specify the number of seconds to wait until index gets ready. If None, wait indefinitely; if >=0, time out after this many seconds; if -1, return immediately and do not wait.
deletion_protection: If enabled, the index cannot be deleted. If disabled, the index can be deleted.
vector_type: The type of vectors to be stored in the index. One of {"dense", "sparse"}.
tags: Tags are key-value pairs you can attach to indexes to better understand, organize, and identify your resources. Some example use cases include tagging indexes with the name of the model that generated the embeddings, the date the index was created, or the purpose of the index.

Returns

A IndexModel instance containing a description of the index that was created.

Creating a serverless index

import os
from pinecone import (
    Pinecone,
    ServerlessSpec,
    CloudProvider,
    AwsRegion,
    Metric,
    DeletionProtection,
    VectorType
)

pc = Pinecone(api_key=os.environ.get("PINECONE_API_KEY"))

pc.create_index(
    name="my_index",
    dimension=1536,
    metric=Metric.COSINE,
    spec=ServerlessSpec(
        cloud=CloudProvider.AWS,
        region=AwsRegion.US_WEST_2
    ),
    deletion_protection=DeletionProtection.DISABLED,
    vector_type=VectorType.DENSE,
    tags={
        "model": "clip",
        "app": "image-search",
        "env": "testing"
    }
)

Creating a pod index

import os
from pinecone import (
    Pinecone,
    PodSpec,
    PodIndexEnvironment,
    PodType,
    Metric,
    DeletionProtection,
    VectorType
)

pc = Pinecone(api_key=os.environ.get("PINECONE_API_KEY"))

pc.create_index(
    name="my_index",
    dimension=1536,
    metric=Metric.COSINE,
    spec=PodSpec(
        environment=PodIndexEnvironment.US_EAST4_GCP,
        pod_type=PodType.P1_X1
    ),
    deletion_protection=DeletionProtection.DISABLED,
    tags={
        "model": "clip",
        "app": "image-search",
        "env": "testing"
    }
)

def create_index_for_model( self, name: str, cloud: Union[pinecone.enums.clouds.CloudProvider, str], region: Union[pinecone.enums.clouds.AwsRegion, pinecone.enums.clouds.GcpRegion, pinecone.enums.clouds.AzureRegion, str], embed: Union[pinecone.models.index_embed.IndexEmbed, pinecone.control.types.create_index_for_model_embed.CreateIndexForModelEmbedTypedDict], tags: Optional[Dict[str, str]] = None, deletion_protection: Union[pinecone.enums.deletion_protection.DeletionProtection, str, NoneType] = <DeletionProtection.DISABLED: 'disabled'>, timeout: Optional[int] = None) -> pinecone.models.index_model.IndexModel: View Source

150    def create_index_for_model(
151        self,
152        name: str,
153        cloud: Union[CloudProvider, str],
154        region: Union[AwsRegion, GcpRegion, AzureRegion, str],
155        embed: Union[IndexEmbed, CreateIndexForModelEmbedTypedDict],
156        tags: Optional[Dict[str, str]] = None,
157        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
158        timeout: Optional[int] = None,
159    ) -> IndexModel:
160        req = PineconeDBControlRequestFactory.create_index_for_model_request(
161            name=name,
162            cloud=cloud,
163            region=region,
164            embed=embed,
165            tags=tags,
166            deletion_protection=deletion_protection,
167        )
168        resp = self.index_api.create_index_for_model(req)
169
170        if timeout == -1:
171            return IndexModel(resp)
172        return self.__poll_describe_index_until_ready(name, timeout)

Parameters

name: The name of the index to create. Must be unique within your project and cannot be changed once created. Allowed characters are lowercase letters, numbers, and hyphens and the name may not begin or end with hyphens. Maximum length is 45 characters.
cloud: The cloud provider to use for the index. One of {"aws", "gcp", "azure"}.
region: The region to use for the index. Enum objects AwsRegion, GcpRegion, and AzureRegion are also available to help you quickly set these parameters, but may not be up to date as new regions become available.
embed: The embedding configuration for the index. This param accepts a dictionary or an instance of the IndexEmbed object.
tags: Tags are key-value pairs you can attach to indexes to better understand, organize, and identify your resources. Some example use cases include tagging indexes with the name of the model that generated the embeddings, the date the index was created, or the purpose of the index.
deletion_protection: If enabled, the index cannot be deleted. If disabled, the index can be deleted. This setting can be changed with configure_index.
timeout: Specify the number of seconds to wait until index is ready to receive data. If None, wait indefinitely; if >=0, time out after this many seconds; if -1, return immediately and do not wait.

Returns

A description of the index that was created.

This method is used to create a Serverless index that is configured for use with Pinecone's integrated inference models.

The resulting index can be described, listed, configured, and deleted like any other Pinecone index with the describe_index, list_indexes, configure_index, and delete_index methods.

After the model is created, you can upsert records into the index with the upsert_records method, and search your records with the search method.

from pinecone import (
    Pinecone,
    IndexEmbed,
    CloudProvider,
    AwsRegion,
    EmbedModel,
    Metric,
)

pc = Pinecone()

if not pc.has_index("book-search"):
    desc = await pc.create_index_for_model(
        name="book-search",
        cloud=CloudProvider.AWS,
        region=AwsRegion.US_EAST_1,
        embed=IndexEmbed(
            model=EmbedModel.Multilingual_E5_Large,
            metric=Metric.COSINE,
            field_map={
                "text": "description",
            },
        )
    )

To see the available cloud regions, see this pinecone.control.pinecone.io/troubleshooting/available-cloud-regions">Pinecone documentation page.

See the pinecone.control.pinecone.io/models/overview">Model Gallery to learn about available models.

def delete_index(self, name: str, timeout: Optional[int] = None): View Source

212    def delete_index(self, name: str, timeout: Optional[int] = None):
213        self.index_api.delete_index(name)
214        self.index_host_store.delete_host(self.config, name)
215
216        if timeout == -1:
217            return
218
219        if timeout is None:
220            while self.has_index(name):
221                time.sleep(5)
222        else:
223            while self.has_index(name) and timeout >= 0:
224                time.sleep(5)
225                timeout -= 5
226        if timeout and timeout < 0:
227            raise (
228                TimeoutError(
229                    "Please call the list_indexes API ({}) to confirm if index is deleted".format(
230                        "https://www.pinecone.io/docs/api/operation/list_indexes/"
231                    )
232                )
233            )

Parameters

name: the name of the index.
timeout: Number of seconds to poll status checking whether the index has been deleted. If None, wait indefinitely; if >=0, time out after this many seconds; if -1, return immediately and do not wait.

Deletes a Pinecone index.

Deleting an index is an irreversible operation. All data in the index will be lost. When you use this command, a request is sent to the Pinecone control plane to delete the index, but the termination is not synchronous because resources take a few moments to be released.

By default the delete_index method will block until polling of the describe_index method shows that the delete operation has completed. If you prefer to return immediately and not wait for the index to be deleted, you can pass timeout=-1 to the method.

After the delete request is submitted, polling describe_index will show that the index transitions into a Terminating state before eventually resulting in a 404 after it has been removed.

This operation can fail if the index is configured with deletion_protection="enabled". In this case, you will need to call configure_index to disable deletion protection before you can delete the index.

from pinecone import Pinecone

pc = Pinecone()

index_name = "my_index"
desc = pc.describe_index(name=index_name)

if desc.deletion_protection == "enabled":
    # If for some reason deletion protection is enabled, you will need to disable it first
    # before you can delete the index. But use caution as this operation is not reversible
    # and if somebody enabled deletion protection, they probably had a good reason.
    pc.configure_index(name=index_name, deletion_protection="disabled")

pc.delete_index(name=index_name)

def list_indexes(self) -> pinecone.models.index_list.IndexList: View Source

235    def list_indexes(self) -> IndexList:
236        response = self.index_api.list_indexes()
237        return IndexList(response)

Returns

Returns an IndexList object, which is iterable and contains a list of IndexModel objects. The IndexList also has a convenience method names() which returns a list of index names for situations where you just want to iterate over all index names.

Lists all indexes in your project.

The results include a description of all indexes in your project, including the index name, dimension, metric, status, and spec.

If you simply want to check whether an index exists, see the has_index() convenience method.

You can use the list_indexes() method to iterate over descriptions of every index in your project.

from pinecone import Pinecone

pc = Pinecone()

for index in pc.list_indexes():
    print(index.name)
    print(index.dimension)
    print(index.metric)
    print(index.status)
    print(index.host)
    print(index.spec)

def describe_index(self, name: str) -> pinecone.models.index_model.IndexModel: View Source

239    def describe_index(self, name: str) -> IndexModel:
240        api_instance = self.index_api
241        description = api_instance.describe_index(name)
242        host = description.host
243        self.index_host_store.set_host(self.config, name, host)
244
245        return IndexModel(description)

Parameters

name: the name of the index to describe.

Returns

Returns an IndexModel object which gives access to properties such as the index name, dimension, metric, host url, status, and spec.

Describes a Pinecone index.

Getting your index host url

In a real production situation, you probably want to store the host url in an environment variable so you don't have to call describe_index and re-fetch it every time you want to use the index. But this example shows how to get the value from the API using describe_index.

from pinecone import Pinecone, Index

pc = Pinecone()

index_name="my_index"
description = pc.describe_index(name=index_name)
print(description)
# {
#     "name": "my_index",
#     "metric": "cosine",
#     "host": "my_index-dojoi3u.svc.aped-4627-b74a.pinecone.io",
#     "spec": {
#         "serverless": {
#             "cloud": "aws",
#             "region": "us-east-1"
#         }
#     },
#     "status": {
#         "ready": true,
#         "state": "Ready"
#     },
#     "vector_type": "dense",
#     "dimension": 1024,
#     "deletion_protection": "enabled",
#     "tags": {
#         "environment": "production"
#     }
# }

print(f"Your index is hosted at {description.host}")

index = pc.Index(host=description.host)
index.upsert(vectors=[...])

def has_index(self, name: str) -> bool: View Source

247    def has_index(self, name: str) -> bool:
248        if name in self.list_indexes().names():
249            return True
250        else:
251            return False

Parameters

name: The name of the index to check for existence.

Returns

Returns True if the index exists, False otherwise.

Checks if a Pinecone index exists.

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone()

index_name = "my_index"
if not pc.has_index(index_name):
    print("Index does not exist, creating...")
    pc.create_index(
        name=index_name,
        dimension=768,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-west-2")
    )

def configure_index( self, name: str, replicas: Optional[int] = None, pod_type: Union[pinecone.enums.pod_type.PodType, str, NoneType] = None, deletion_protection: Union[pinecone.enums.deletion_protection.DeletionProtection, str, NoneType] = None, tags: Optional[Dict[str, str]] = None): View Source

253    def configure_index(
254        self,
255        name: str,
256        replicas: Optional[int] = None,
257        pod_type: Optional[Union[PodType, str]] = None,
258        deletion_protection: Optional[Union[DeletionProtection, str]] = None,
259        tags: Optional[Dict[str, str]] = None,
260    ):
261        api_instance = self.index_api
262        description = self.describe_index(name=name)
263
264        req = PineconeDBControlRequestFactory.configure_index_request(
265            description=description,
266            replicas=replicas,
267            pod_type=pod_type,
268            deletion_protection=deletion_protection,
269            tags=tags,
270        )
271        api_instance.configure_index(name, configure_index_request=req)

Parameters

name: the name of the Index
replicas: the desired number of replicas, lowest value is 0.
pod_type: the new pod_type for the index. To learn more about the available pod types, please see pinecone.control.pinecone.io/docs/indexes">Understanding Indexes
deletion_protection: If set to 'enabled', the index cannot be deleted. If 'disabled', the index can be deleted.
tags: A dictionary of tags to apply to the index. Tags are key-value pairs that can be used to organize and manage indexes. To remove a tag, set the value to "". Tags passed to configure_index will be merged with existing tags and any with the value empty string will be removed.

This method is used to modify an index's configuration. It can be used to:

Scale a pod-based index horizontally using replicas
Scale a pod-based index vertically using pod_type
Enable or disable deletion protection using deletion_protection
Add, change, or remove tags using tags

Scaling pod-based indexes

To scale your pod-based index, you pass a replicas and/or pod_type param to the configure_index method. pod_type may be a string or a value from the PodType enum.

from pinecone import Pinecone, PodType

pc = Pinecone()
pc.configure_index(
    name="my_index",
    replicas=2,
    pod_type=PodType.P1_X2
)

After providing these new configurations, you must call describe_index to see the status of the index as the changes are applied.

Enabling or disabling deletion protection

To enable or disable deletion protection, pass the deletion_protection parameter to the configure_index method. When deletion protection is enabled, the index cannot be deleted with the delete_index method.

from pinecone import Pinecone, DeletionProtection

pc = Pinecone()

# Enable deletion protection
pc.configure_index(
    name="my_index",
    deletion_protection=DeletionProtection.ENABLED
)

# Call describe_index to see the change was applied.
assert pc.describe_index("my_index").deletion_protection == "enabled"

# Disable deletion protection
pc.configure_index(
    name="my_index",
    deletion_protection=DeletionProtection.DISABLED
)

Adding, changing, or removing tags

To add, change, or remove tags, pass the tags parameter to the configure_index method. When tags are passed using configure_index, they are merged with any existing tags already on the index. To remove a tag, set the value of the key to an empty string.

from pinecone import Pinecone

pc = Pinecone()

# Add a tag
pc.configure_index(name="my_index", tags={"environment": "staging"})

# Change a tag
pc.configure_index(name="my_index", tags={"environment": "production"})

# Remove a tag
pc.configure_index(name="my_index", tags={"environment": ""})

# Call describe_index to view the tags are changed
print(pc.describe_index("my_index").tags)

def create_collection(self, name: str, source: str) -> None: View Source

273    def create_collection(self, name: str, source: str) -> None:
274        req = PineconeDBControlRequestFactory.create_collection_request(name=name, source=source)
275        self.index_api.create_collection(create_collection_request=req)

Create a collection from a pod-based index

Parameters

name: Name of the collection
source: Name of the source index

def list_collections(self) -> pinecone.models.collection_list.CollectionList: View Source

277    def list_collections(self) -> CollectionList:
278        response = self.index_api.list_collections()
279        return CollectionList(response)

List all collections

from pinecone import Pinecone

pc = Pinecone()

for collection in pc.list_collections():
    print(collection.name)
    print(collection.source)

# You can also iterate specifically over the collection
# names with the .names() helper.
collection_name="my_collection"
for collection_name in pc.list_collections().names():
    print(collection_name)

def delete_collection(self, name: str) -> None: View Source

281    def delete_collection(self, name: str) -> None:
282        self.index_api.delete_collection(name)

Parameters

name: The name of the collection to delete.

Deletes a collection.

Deleting a collection is an irreversible operation. All data in the collection will be lost.

This method tells Pinecone you would like to delete a collection, but it takes a few moments to complete the operation. Use the describe_collection() method to confirm that the collection has been deleted.

from pinecone import Pinecone

pc = Pinecone()

pc.delete_collection(name="my_collection")

def describe_collection(self, name: str): View Source

284    def describe_collection(self, name: str):
285        return self.index_api.describe_collection(name).to_dict()

Describes a collection.

Parameters

The name of the collection

Returns

Description of the collection

from pinecone import Pinecone

pc = Pinecone()

description = pc.describe_collection("my_collection")
print(description.name)
print(description.source)
print(description.status)
print(description.size)

def Index(self, name: str = '', host: str = '', **kwargs): View Source

297    def Index(self, name: str = "", host: str = "", **kwargs):
298        if name == "" and host == "":
299            raise ValueError("Either name or host must be specified")
300
301        pt = kwargs.pop("pool_threads", None) or self.pool_threads
302        api_key = self.config.api_key
303        openapi_config = self.openapi_config
304
305        if host != "":
306            check_realistic_host(host)
307
308            # Use host url if it is provided
309            index_host = normalize_host(host)
310        else:
311            # Otherwise, get host url from describe_index using the index name
312            index_host = self.index_host_store.get_host(self.index_api, self.config, name)
313
314        return _Index(
315            host=index_host,
316            api_key=api_key,
317            pool_threads=pt,
318            openapi_config=openapi_config,
319            source_tag=self.config.source_tag,
320            **kwargs,
321        )

Parameters

name: The name of the index to target. If you specify the name of the index, the client will fetch the host url from the Pinecone control plane.
host: The host url of the index to target. If you specify the host url, the client will use the host url directly without making any additional calls to the control plane.
pool_threads: The number of threads to use when making parallel requests by calling index methods with optional kwarg async_req=True, or using methods that make use of thread-based parallelism automatically such as query_namespaces().
connection_pool_maxsize: The maximum number of connections to keep in the connection pool.

Returns

An instance of the Index class.

Target an index for data operations.

Target an index by host url

In production situations, you want to uspert or query your data as quickly as possible. If you know in advance the host url of your index, you can eliminate a round trip to the Pinecone control plane by specifying the host of the index. If instead you pass the name of the index, the client will need to make an additional call to api.pinecone.io to get the host url before any data operations can take place.

import os
from pinecone import Pinecone

api_key = os.environ.get("PINECONE_API_KEY")
index_host = os.environ.get("PINECONE_INDEX_HOST")

pc = Pinecone(api_key=api_key)
index = pc.Index(host=index_host)

# Now you're ready to perform data operations
index.query(vector=[...], top_k=10)

To find your host url, you can use the describe_index method to call api.pinecone.io. The host url is returned in the response. Or, alternatively, the host is displayed in the Pinecone web console.

import os
from pinecone import Pinecone

pc = Pinecone(
    api_key=os.environ.get("PINECONE_API_KEY")
)

host = pc.describe_index('index-name').host

Target an index by name (not recommended for production)

For more casual usage, such as when you are playing and exploring with Pinecone in a notebook setting, you can also target an index by name. If you use this approach, the client may need to perform an extra call to the Pinecone control plane to get the host url on your behalf to get the index host.

The client will cache the index host for future use whenever it is seen, so you will only incur the overhead of only one call. But this approach is not recommended for production usage because it introduces an unnecessary runtime dependency on api.pinecone.io.

import os
from pinecone import Pinecone, ServerlessSpec

api_key = os.environ.get("PINECONE_API_KEY")

pc = Pinecone(api_key=api_key)
pc.create_index(
    name='my_index',
    dimension=1536,
    metric='cosine',
    spec=ServerlessSpec(cloud='aws', region='us-west-2')
)
index = pc.Index('my_index')

# Now you're ready to perform data operations
index.query(vector=[...], top_k=10)

def IndexAsyncio(self, host: str, **kwargs): View Source

323    def IndexAsyncio(self, host: str, **kwargs):
324        api_key = self.config.api_key
325        openapi_config = self.openapi_config
326
327        if host is None or host == "":
328            raise ValueError("A host must be specified")
329
330        check_realistic_host(host)
331        index_host = normalize_host(host)
332
333        return _IndexAsyncio(
334            host=index_host,
335            api_key=api_key,
336            openapi_config=openapi_config,
337            source_tag=self.config.source_tag,
338            **kwargs,
339        )

Build an asyncio-compatible Index object.

Parameters

host: The host url of the index to target. You can find this url in the Pinecone web console or by calling describe_index method of Pinecone or PineconeAsyncio.

Returns

An instance of the IndexAsyncio class.