pinecone.control.pinecone_asyncio 

  1import logging
  2import asyncio
  3from typing import Optional, Dict, Union
  4
  5from pinecone.config import PineconeConfig, ConfigBuilder
  6
  7from pinecone.core.openapi.db_control.api.manage_indexes_api import AsyncioManageIndexesApi
  8from pinecone.openapi_support import AsyncioApiClient
  9
 10from pinecone.utils import normalize_host, setup_async_openapi_client
 11from pinecone.core.openapi.db_control import API_VERSION
 12from pinecone.models import (
 13    ServerlessSpec,
 14    PodSpec,
 15    IndexModel,
 16    IndexList,
 17    CollectionList,
 18    IndexEmbed,
 19)
 20from pinecone.utils import docslinks
 21
 22from pinecone.data import _IndexAsyncio, _AsyncioInference
 23from pinecone.enums import (
 24    Metric,
 25    VectorType,
 26    DeletionProtection,
 27    PodType,
 28    CloudProvider,
 29    AwsRegion,
 30    GcpRegion,
 31    AzureRegion,
 32)
 33from .types import CreateIndexForModelEmbedTypedDict
 34from .request_factory import PineconeDBControlRequestFactory
 35from .pinecone_interface_asyncio import PineconeAsyncioDBControlInterface
 36from .pinecone import check_realistic_host
 37
 38logger = logging.getLogger(__name__)
 39""" @private """
 40
 41
 42class PineconeAsyncio(PineconeAsyncioDBControlInterface):
 43    """
 44    `PineconeAsyncio` is an asyncio client for interacting with Pinecone's control plane API.
 45
 46    This class implements methods for managing and interacting with Pinecone resources
 47    such as collections and indexes.
 48
 49    To perform data operations such as inserting and querying vectors, use the `IndexAsyncio` class.
 50
 51    ```python
 52    import asyncio
 53    from pinecone import Pinecone
 54
 55    async def main():
 56        pc = Pinecone()
 57        async with pc.IndexAsyncio(host="my-index.pinecone.io") as idx:
 58            await idx.upsert(vectors=[(1, [1, 2, 3]), (2, [4, 5, 6])])
 59
 60    asyncio.run(main())
 61    ```
 62    """
 63
 64    def __init__(
 65        self,
 66        api_key: Optional[str] = None,
 67        host: Optional[str] = None,
 68        proxy_url: Optional[str] = None,
 69        # proxy_headers: Optional[Dict[str, str]] = None,
 70        ssl_ca_certs: Optional[str] = None,
 71        ssl_verify: Optional[bool] = None,
 72        additional_headers: Optional[Dict[str, str]] = {},
 73        **kwargs,
 74    ):
 75        for deprecated_kwarg in {"config", "openapi_config"}:
 76            if deprecated_kwarg in kwargs:
 77                raise NotImplementedError(
 78                    f"Passing {deprecated_kwarg} is no longer supported. Please pass individual settings such as proxy_url, ssl_ca_certs, and ssl_verify directly to the Pinecone constructor as keyword arguments. See the README at {docslinks['README']} for examples."
 79                )
 80
 81        for unimplemented_kwarg in {"proxy_headers"}:
 82            if unimplemented_kwarg in kwargs:
 83                raise NotImplementedError(
 84                    f"You have passed {unimplemented_kwarg} but this configuration has not been implemented for PineconeAsyncio."
 85                )
 86
 87        self.config = PineconeConfig.build(
 88            api_key=api_key,
 89            host=host,
 90            additional_headers=additional_headers,
 91            proxy_url=proxy_url,
 92            proxy_headers=None,
 93            ssl_ca_certs=ssl_ca_certs,
 94            ssl_verify=ssl_verify,
 95            **kwargs,
 96        )
 97        """ @private """
 98
 99        self.openapi_config = ConfigBuilder.build_openapi_config(self.config, **kwargs)
100        """ @private """
101
102        self._inference = None  # Lazy initialization
103        """ @private """
104
105        self.index_api = setup_async_openapi_client(
106            api_client_klass=AsyncioApiClient,
107            api_klass=AsyncioManageIndexesApi,
108            config=self.config,
109            openapi_config=self.openapi_config,
110            api_version=API_VERSION,
111        )
112        """ @private """
113
114    async def __aenter__(self):
115        return self
116
117    async def __aexit__(self, exc_type, exc_value, traceback):
118        await self.close()
119
120    async def close(self):
121        """Cleanup resources used by the Pinecone client.
122
123        This method should be called when the client is no longer needed so that
124        it can cleanup the aioahttp session and other resources.
125
126        After close has been called, the client instance should not be used.
127
128        ```python
129        import asyncio
130        from pinecone import PineconeAsyncio
131
132        async def main():
133            pc = PineconeAsyncio()
134            desc = await pc.describe_index(name="my-index")
135            await pc.close()
136
137        asyncio.run(main())
138        ```
139
140        If you are using the client as a context manager, the close method is called automatically
141        when exiting.
142
143        ```python
144        import asyncio
145        from pinecone import PineconeAsyncio
146
147        async def main():
148            async with PineconeAsyncio() as pc:
149                desc = await pc.describe_index(name="my-index")
150
151        # No need to call close in this case because the "async with" syntax
152        # automatically calls close when exiting the block.
153        asyncio.run(main())
154        ```
155
156        """
157        await self.index_api.api_client.close()
158
159    @property
160    def inference(self):
161        """Dynamically create and cache the Inference instance."""
162        if self._inference is None:
163            self._inference = _AsyncioInference(api_client=self.index_api.api_client)
164        return self._inference
165
166    async def create_index(
167        self,
168        name: str,
169        spec: Union[Dict, ServerlessSpec, PodSpec],
170        dimension: Optional[int] = None,
171        metric: Optional[Union[Metric, str]] = Metric.COSINE,
172        timeout: Optional[int] = None,
173        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
174        vector_type: Optional[Union[VectorType, str]] = VectorType.DENSE,
175        tags: Optional[Dict[str, str]] = None,
176    ) -> IndexModel:
177        req = PineconeDBControlRequestFactory.create_index_request(
178            name=name,
179            spec=spec,
180            dimension=dimension,
181            metric=metric,
182            deletion_protection=deletion_protection,
183            vector_type=vector_type,
184            tags=tags,
185        )
186        resp = await self.index_api.create_index(create_index_request=req)
187
188        if timeout == -1:
189            return IndexModel(resp)
190        return await self.__poll_describe_index_until_ready(name, timeout)
191
192    async def create_index_for_model(
193        self,
194        name: str,
195        cloud: Union[CloudProvider, str],
196        region: Union[AwsRegion, GcpRegion, AzureRegion, str],
197        embed: Union[IndexEmbed, CreateIndexForModelEmbedTypedDict],
198        tags: Optional[Dict[str, str]] = None,
199        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
200        timeout: Optional[int] = None,
201    ) -> IndexModel:
202        req = PineconeDBControlRequestFactory.create_index_for_model_request(
203            name=name,
204            cloud=cloud,
205            region=region,
206            embed=embed,
207            tags=tags,
208            deletion_protection=deletion_protection,
209        )
210        resp = await self.index_api.create_index_for_model(req)
211
212        if timeout == -1:
213            return IndexModel(resp)
214        return await self.__poll_describe_index_until_ready(name, timeout)
215
216    async def __poll_describe_index_until_ready(self, name: str, timeout: Optional[int] = None):
217        description = None
218
219        async def is_ready() -> bool:
220            nonlocal description
221            description = await self.describe_index(name=name)
222            return description.status.ready
223
224        total_wait_time = 0
225        if timeout is None:
226            # Wait indefinitely
227            while not await is_ready():
228                logger.debug(
229                    f"Waiting for index {name} to be ready. Total wait time {total_wait_time} seconds."
230                )
231                total_wait_time += 5
232                await asyncio.sleep(5)
233
234        else:
235            # Wait for a maximum of timeout seconds
236            while not await is_ready():
237                if timeout < 0:
238                    logger.error(f"Index {name} is not ready. Timeout reached.")
239                    link = docslinks["API_DESCRIBE_INDEX"]
240                    timeout_msg = (
241                        f"Please call describe_index() to confirm index status. See docs at {link}"
242                    )
243                    raise TimeoutError(timeout_msg)
244
245                logger.debug(
246                    f"Waiting for index {name} to be ready. Total wait time: {total_wait_time}"
247                )
248                total_wait_time += 5
249                await asyncio.sleep(5)
250                timeout -= 5
251
252        return description
253
254    async def delete_index(self, name: str, timeout: Optional[int] = None):
255        await self.index_api.delete_index(name)
256
257        if timeout == -1:
258            return
259
260        if timeout is None:
261            while await self.has_index(name):
262                await asyncio.sleep(5)
263        else:
264            while await self.has_index(name) and timeout >= 0:
265                await asyncio.sleep(5)
266                timeout -= 5
267        if timeout and timeout < 0:
268            raise (
269                TimeoutError(
270                    "Please call the list_indexes API ({}) to confirm if index is deleted".format(
271                        "https://www.pinecone.io/docs/api/operation/list_indexes/"
272                    )
273                )
274            )
275
276    async def list_indexes(self) -> IndexList:
277        response = await self.index_api.list_indexes()
278        return IndexList(response)
279
280    async def describe_index(self, name: str) -> IndexModel:
281        description = await self.index_api.describe_index(name)
282        return IndexModel(description)
283
284    async def has_index(self, name: str) -> bool:
285        available_indexes = await self.list_indexes()
286        if name in available_indexes.names():
287            return True
288        else:
289            return False
290
291    async def configure_index(
292        self,
293        name: str,
294        replicas: Optional[int] = None,
295        pod_type: Optional[Union[PodType, str]] = None,
296        deletion_protection: Optional[Union[DeletionProtection, str]] = None,
297        tags: Optional[Dict[str, str]] = None,
298    ):
299        description = await self.describe_index(name=name)
300
301        req = PineconeDBControlRequestFactory.configure_index_request(
302            description=description,
303            replicas=replicas,
304            pod_type=pod_type,
305            deletion_protection=deletion_protection,
306            tags=tags,
307        )
308        await self.index_api.configure_index(name, configure_index_request=req)
309
310    async def create_collection(self, name: str, source: str):
311        req = PineconeDBControlRequestFactory.create_collection_request(name=name, source=source)
312        await self.index_api.create_collection(create_collection_request=req)
313
314    async def list_collections(self) -> CollectionList:
315        response = await self.index_api.list_collections()
316        return CollectionList(response)
317
318    async def delete_collection(self, name: str):
319        await self.index_api.delete_collection(name)
320
321    async def describe_collection(self, name: str):
322        return await self.index_api.describe_collection(name).to_dict()
323
324    def IndexAsyncio(self, host: str, **kwargs) -> _IndexAsyncio:
325        api_key = self.config.api_key
326        openapi_config = self.openapi_config
327
328        if host is None or host == "":
329            raise ValueError("A host must be specified")
330
331        check_realistic_host(host)
332        index_host = normalize_host(host)
333
334        return _IndexAsyncio(
335            host=index_host,
336            api_key=api_key,
337            openapi_config=openapi_config,
338            source_tag=self.config.source_tag,
339            **kwargs,
340        )
class PineconeAsyncio(pinecone.control.pinecone_interface_asyncio.PineconeAsyncioDBControlInterface): 
 43class PineconeAsyncio(PineconeAsyncioDBControlInterface):
 44    """
 45    `PineconeAsyncio` is an asyncio client for interacting with Pinecone's control plane API.
 46
 47    This class implements methods for managing and interacting with Pinecone resources
 48    such as collections and indexes.
 49
 50    To perform data operations such as inserting and querying vectors, use the `IndexAsyncio` class.
 51
 52    ```python
 53    import asyncio
 54    from pinecone import Pinecone
 55
 56    async def main():
 57        pc = Pinecone()
 58        async with pc.IndexAsyncio(host="my-index.pinecone.io") as idx:
 59            await idx.upsert(vectors=[(1, [1, 2, 3]), (2, [4, 5, 6])])
 60
 61    asyncio.run(main())
 62    ```
 63    """
 64
 65    def __init__(
 66        self,
 67        api_key: Optional[str] = None,
 68        host: Optional[str] = None,
 69        proxy_url: Optional[str] = None,
 70        # proxy_headers: Optional[Dict[str, str]] = None,
 71        ssl_ca_certs: Optional[str] = None,
 72        ssl_verify: Optional[bool] = None,
 73        additional_headers: Optional[Dict[str, str]] = {},
 74        **kwargs,
 75    ):
 76        for deprecated_kwarg in {"config", "openapi_config"}:
 77            if deprecated_kwarg in kwargs:
 78                raise NotImplementedError(
 79                    f"Passing {deprecated_kwarg} is no longer supported. Please pass individual settings such as proxy_url, ssl_ca_certs, and ssl_verify directly to the Pinecone constructor as keyword arguments. See the README at {docslinks['README']} for examples."
 80                )
 81
 82        for unimplemented_kwarg in {"proxy_headers"}:
 83            if unimplemented_kwarg in kwargs:
 84                raise NotImplementedError(
 85                    f"You have passed {unimplemented_kwarg} but this configuration has not been implemented for PineconeAsyncio."
 86                )
 87
 88        self.config = PineconeConfig.build(
 89            api_key=api_key,
 90            host=host,
 91            additional_headers=additional_headers,
 92            proxy_url=proxy_url,
 93            proxy_headers=None,
 94            ssl_ca_certs=ssl_ca_certs,
 95            ssl_verify=ssl_verify,
 96            **kwargs,
 97        )
 98        """ @private """
 99
100        self.openapi_config = ConfigBuilder.build_openapi_config(self.config, **kwargs)
101        """ @private """
102
103        self._inference = None  # Lazy initialization
104        """ @private """
105
106        self.index_api = setup_async_openapi_client(
107            api_client_klass=AsyncioApiClient,
108            api_klass=AsyncioManageIndexesApi,
109            config=self.config,
110            openapi_config=self.openapi_config,
111            api_version=API_VERSION,
112        )
113        """ @private """
114
115    async def __aenter__(self):
116        return self
117
118    async def __aexit__(self, exc_type, exc_value, traceback):
119        await self.close()
120
121    async def close(self):
122        """Cleanup resources used by the Pinecone client.
123
124        This method should be called when the client is no longer needed so that
125        it can cleanup the aioahttp session and other resources.
126
127        After close has been called, the client instance should not be used.
128
129        ```python
130        import asyncio
131        from pinecone import PineconeAsyncio
132
133        async def main():
134            pc = PineconeAsyncio()
135            desc = await pc.describe_index(name="my-index")
136            await pc.close()
137
138        asyncio.run(main())
139        ```
140
141        If you are using the client as a context manager, the close method is called automatically
142        when exiting.
143
144        ```python
145        import asyncio
146        from pinecone import PineconeAsyncio
147
148        async def main():
149            async with PineconeAsyncio() as pc:
150                desc = await pc.describe_index(name="my-index")
151
152        # No need to call close in this case because the "async with" syntax
153        # automatically calls close when exiting the block.
154        asyncio.run(main())
155        ```
156
157        """
158        await self.index_api.api_client.close()
159
160    @property
161    def inference(self):
162        """Dynamically create and cache the Inference instance."""
163        if self._inference is None:
164            self._inference = _AsyncioInference(api_client=self.index_api.api_client)
165        return self._inference
166
167    async def create_index(
168        self,
169        name: str,
170        spec: Union[Dict, ServerlessSpec, PodSpec],
171        dimension: Optional[int] = None,
172        metric: Optional[Union[Metric, str]] = Metric.COSINE,
173        timeout: Optional[int] = None,
174        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
175        vector_type: Optional[Union[VectorType, str]] = VectorType.DENSE,
176        tags: Optional[Dict[str, str]] = None,
177    ) -> IndexModel:
178        req = PineconeDBControlRequestFactory.create_index_request(
179            name=name,
180            spec=spec,
181            dimension=dimension,
182            metric=metric,
183            deletion_protection=deletion_protection,
184            vector_type=vector_type,
185            tags=tags,
186        )
187        resp = await self.index_api.create_index(create_index_request=req)
188
189        if timeout == -1:
190            return IndexModel(resp)
191        return await self.__poll_describe_index_until_ready(name, timeout)
192
193    async def create_index_for_model(
194        self,
195        name: str,
196        cloud: Union[CloudProvider, str],
197        region: Union[AwsRegion, GcpRegion, AzureRegion, str],
198        embed: Union[IndexEmbed, CreateIndexForModelEmbedTypedDict],
199        tags: Optional[Dict[str, str]] = None,
200        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
201        timeout: Optional[int] = None,
202    ) -> IndexModel:
203        req = PineconeDBControlRequestFactory.create_index_for_model_request(
204            name=name,
205            cloud=cloud,
206            region=region,
207            embed=embed,
208            tags=tags,
209            deletion_protection=deletion_protection,
210        )
211        resp = await self.index_api.create_index_for_model(req)
212
213        if timeout == -1:
214            return IndexModel(resp)
215        return await self.__poll_describe_index_until_ready(name, timeout)
216
217    async def __poll_describe_index_until_ready(self, name: str, timeout: Optional[int] = None):
218        description = None
219
220        async def is_ready() -> bool:
221            nonlocal description
222            description = await self.describe_index(name=name)
223            return description.status.ready
224
225        total_wait_time = 0
226        if timeout is None:
227            # Wait indefinitely
228            while not await is_ready():
229                logger.debug(
230                    f"Waiting for index {name} to be ready. Total wait time {total_wait_time} seconds."
231                )
232                total_wait_time += 5
233                await asyncio.sleep(5)
234
235        else:
236            # Wait for a maximum of timeout seconds
237            while not await is_ready():
238                if timeout < 0:
239                    logger.error(f"Index {name} is not ready. Timeout reached.")
240                    link = docslinks["API_DESCRIBE_INDEX"]
241                    timeout_msg = (
242                        f"Please call describe_index() to confirm index status. See docs at {link}"
243                    )
244                    raise TimeoutError(timeout_msg)
245
246                logger.debug(
247                    f"Waiting for index {name} to be ready. Total wait time: {total_wait_time}"
248                )
249                total_wait_time += 5
250                await asyncio.sleep(5)
251                timeout -= 5
252
253        return description
254
255    async def delete_index(self, name: str, timeout: Optional[int] = None):
256        await self.index_api.delete_index(name)
257
258        if timeout == -1:
259            return
260
261        if timeout is None:
262            while await self.has_index(name):
263                await asyncio.sleep(5)
264        else:
265            while await self.has_index(name) and timeout >= 0:
266                await asyncio.sleep(5)
267                timeout -= 5
268        if timeout and timeout < 0:
269            raise (
270                TimeoutError(
271                    "Please call the list_indexes API ({}) to confirm if index is deleted".format(
272                        "https://www.pinecone.io/docs/api/operation/list_indexes/"
273                    )
274                )
275            )
276
277    async def list_indexes(self) -> IndexList:
278        response = await self.index_api.list_indexes()
279        return IndexList(response)
280
281    async def describe_index(self, name: str) -> IndexModel:
282        description = await self.index_api.describe_index(name)
283        return IndexModel(description)
284
285    async def has_index(self, name: str) -> bool:
286        available_indexes = await self.list_indexes()
287        if name in available_indexes.names():
288            return True
289        else:
290            return False
291
292    async def configure_index(
293        self,
294        name: str,
295        replicas: Optional[int] = None,
296        pod_type: Optional[Union[PodType, str]] = None,
297        deletion_protection: Optional[Union[DeletionProtection, str]] = None,
298        tags: Optional[Dict[str, str]] = None,
299    ):
300        description = await self.describe_index(name=name)
301
302        req = PineconeDBControlRequestFactory.configure_index_request(
303            description=description,
304            replicas=replicas,
305            pod_type=pod_type,
306            deletion_protection=deletion_protection,
307            tags=tags,
308        )
309        await self.index_api.configure_index(name, configure_index_request=req)
310
311    async def create_collection(self, name: str, source: str):
312        req = PineconeDBControlRequestFactory.create_collection_request(name=name, source=source)
313        await self.index_api.create_collection(create_collection_request=req)
314
315    async def list_collections(self) -> CollectionList:
316        response = await self.index_api.list_collections()
317        return CollectionList(response)
318
319    async def delete_collection(self, name: str):
320        await self.index_api.delete_collection(name)
321
322    async def describe_collection(self, name: str):
323        return await self.index_api.describe_collection(name).to_dict()
324
325    def IndexAsyncio(self, host: str, **kwargs) -> _IndexAsyncio:
326        api_key = self.config.api_key
327        openapi_config = self.openapi_config
328
329        if host is None or host == "":
330            raise ValueError("A host must be specified")
331
332        check_realistic_host(host)
333        index_host = normalize_host(host)
334
335        return _IndexAsyncio(
336            host=index_host,
337            api_key=api_key,
338            openapi_config=openapi_config,
339            source_tag=self.config.source_tag,
340            **kwargs,
341        )

PineconeAsyncio is an asyncio client for interacting with Pinecone's control plane API.

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

To perform data operations such as inserting and querying vectors, use the IndexAsyncio class.

import asyncio
from pinecone import Pinecone

async def main():
    pc = Pinecone()
    async with pc.IndexAsyncio(host="my-index.pinecone.io") as idx:
        await idx.upsert(vectors=[(1, [1, 2, 3]), (2, [4, 5, 6])])

asyncio.run(main())
PineconeAsyncio( api_key: Optional[str] = None, host: Optional[str] = None, proxy_url: Optional[str] = None, ssl_ca_certs: Optional[str] = None, ssl_verify: Optional[bool] = None, additional_headers: Optional[Dict[str, str]] = {}, **kwargs) 
 65    def __init__(
 66        self,
 67        api_key: Optional[str] = None,
 68        host: Optional[str] = None,
 69        proxy_url: Optional[str] = None,
 70        # proxy_headers: Optional[Dict[str, str]] = None,
 71        ssl_ca_certs: Optional[str] = None,
 72        ssl_verify: Optional[bool] = None,
 73        additional_headers: Optional[Dict[str, str]] = {},
 74        **kwargs,
 75    ):
 76        for deprecated_kwarg in {"config", "openapi_config"}:
 77            if deprecated_kwarg in kwargs:
 78                raise NotImplementedError(
 79                    f"Passing {deprecated_kwarg} is no longer supported. Please pass individual settings such as proxy_url, ssl_ca_certs, and ssl_verify directly to the Pinecone constructor as keyword arguments. See the README at {docslinks['README']} for examples."
 80                )
 81
 82        for unimplemented_kwarg in {"proxy_headers"}:
 83            if unimplemented_kwarg in kwargs:
 84                raise NotImplementedError(
 85                    f"You have passed {unimplemented_kwarg} but this configuration has not been implemented for PineconeAsyncio."
 86                )
 87
 88        self.config = PineconeConfig.build(
 89            api_key=api_key,
 90            host=host,
 91            additional_headers=additional_headers,
 92            proxy_url=proxy_url,
 93            proxy_headers=None,
 94            ssl_ca_certs=ssl_ca_certs,
 95            ssl_verify=ssl_verify,
 96            **kwargs,
 97        )
 98        """ @private """
 99
100        self.openapi_config = ConfigBuilder.build_openapi_config(self.config, **kwargs)
101        """ @private """
102
103        self._inference = None  # Lazy initialization
104        """ @private """
105
106        self.index_api = setup_async_openapi_client(
107            api_client_klass=AsyncioApiClient,
108            api_klass=AsyncioManageIndexesApi,
109            config=self.config,
110            openapi_config=self.openapi_config,
111            api_version=API_VERSION,
112        )
113        """ @private """

The PineconeAsyncio class is the main entry point for interacting with Pinecone using asyncio. It is used to create, delete, and manage your indexes and collections. Except for needing to use async with when instantiating the client and await when calling its methods, the functionality provided by this class is extremely similar to the functionality of the Pinecone class.

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 to connect to.
  • proxy_url: The URL of the proxy to use for the connection. Default: None
  • proxy_headers: Additional headers to pass to the proxy. Use this if your proxy setup requires authentication. Default: {}
  • 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. Default: None
  • ssl_verify: SSL verification is performed by default, but can be disabled using the boolean flag. Default: True
  • config: A pinecone.config.Config object. If passed, the api_key and host parameters will be ignored.
  • additional_headers: Additional headers to pass to the API. Default: {}

Managing the async context

The PineconeAsyncio class relies on an underlying aiohttp ClientSession to make asynchronous HTTP requests. To ensure that the session is properly closed, you should use the async with syntax when creating a PineconeAsyncio object. This will ensure that the session is properly closed when the context is exited.

import asyncio
from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio(api_key='YOUR_API_KEY') as pc:
        # Do async things
        index_list = await pc.list_indexes()

asyncio.run(main())

As an alternative, if you prefer to avoid code with a nested appearance and are willing to manage cleanup yourself, you can await the close() method to close the session when you are done.

import asyncio
from pinecone import PineconeAsyncio

async def main():
    pc = PineconeAsyncio(api_key='YOUR_API_KEY')

    # Do async things
    index_list = await pc.list_indexes()

    # You're responsible for calling this yourself
    await pc.close()

asyncio.run(main())

Failing to do this may result in error messages appearing from the underlyling aiohttp library.

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.

import asyncio
from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio() as pc:
        # Do async things
        index_list = await pc.list_indexes()

asyncio.run(main())

Configuration with keyword arguments

If you prefer being more explicit in your code, you can also pass the API as

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.

import asyncio
from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio() as pc:
        # Do async things
        index_list = await pc.list_indexes()

asyncio.run(main())

Configuration with keyword arguments

If you prefer being more explicit in your code, you can also pass the API as

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.

import asyncio
from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio() as pc:
        # Do async things
        index_list = await pc.list_indexes()

asyncio.run(main())

Configuration with keyword arguments

If you prefer being more explicit in your code, you can also pass the API as a keyword argument.

import os
import asyncio
from pinecone import PineconeAsyncio

async def main():
    async with Pinecone(api_key=os.environ.get("PINECONE_API_KEY")) as pc:
        # Do async things
        index_list = await pc.list_indexes()

asyncio.run(main())

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.

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 used to configure an SSL context and passed to aiohttp, which is the underlying library currently used by the PineconeAsyncio client to make HTTP requests.

Here is a basic example:

import asyncio
from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio(
        api_key='YOUR_API_KEY',
        proxy_url='https://your-proxy.com'
    ) as pc:
        # Do async things
        index_list = await pc.list_indexes()

asyncio.run(main())

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.

import asyncio
from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio(
        api_key='YOUR_API_KEY',
        proxy_url='https://your-proxy.com',
        ssl_ca_certs='path/to/cert-bundle.pem'
    ) as pc:
        # Do async things
        await pc.list_indexes()

asyncio.run(main())

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 but there are situations where this is useful such as testing with Pinecone Local running in a docker container.

import asyncio
from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio(
        api_key='YOUR_API_KEY',
        ssl_verify=False
    ) as pc:
        if not await pc.has_index('my_index'):
            await pc.create_index(
                name='my_index',
                dimension=1536,
                metric='cosine',
                spec=ServerlessSpec(cloud='aws', region='us-west-2')
            )

asyncio.run(main())

Passing additional headers

If you need to pass additional headers with each request to the Pinecone API, you can do so using the additional_headers parameter. This is primarily for internal testing and end-users shouldn't need to do this unless specifically instructed to do so.

import asyncio
from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio(
        api_key='YOUR_API_KEY',
        host='https://api-staging.pinecone.io',
        additional_headers={'X-My-Header': 'my-value'}
    ) as pc:
        # Do async things
        await pc.list_indexes()

asyncio.run(main())
async def close(self): 
121    async def close(self):
122        """Cleanup resources used by the Pinecone client.
123
124        This method should be called when the client is no longer needed so that
125        it can cleanup the aioahttp session and other resources.
126
127        After close has been called, the client instance should not be used.
128
129        ```python
130        import asyncio
131        from pinecone import PineconeAsyncio
132
133        async def main():
134            pc = PineconeAsyncio()
135            desc = await pc.describe_index(name="my-index")
136            await pc.close()
137
138        asyncio.run(main())
139        ```
140
141        If you are using the client as a context manager, the close method is called automatically
142        when exiting.
143
144        ```python
145        import asyncio
146        from pinecone import PineconeAsyncio
147
148        async def main():
149            async with PineconeAsyncio() as pc:
150                desc = await pc.describe_index(name="my-index")
151
152        # No need to call close in this case because the "async with" syntax
153        # automatically calls close when exiting the block.
154        asyncio.run(main())
155        ```
156
157        """
158        await self.index_api.api_client.close()

Cleanup resources used by the Pinecone client.

This method should be called when the client is no longer needed so that it can cleanup the aioahttp session and other resources.

After close has been called, the client instance should not be used.

import asyncio
from pinecone import PineconeAsyncio

async def main():
    pc = PineconeAsyncio()
    desc = await pc.describe_index(name="my-index")
    await pc.close()

asyncio.run(main())

If you are using the client as a context manager, the close method is called automatically when exiting.

import asyncio
from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio() as pc:
        desc = await pc.describe_index(name="my-index")

# No need to call close in this case because the "async with" syntax
# automatically calls close when exiting the block.
asyncio.run(main())
inference 
160    @property
161    def inference(self):
162        """Dynamically create and cache the Inference instance."""
163        if self._inference is None:
164            self._inference = _AsyncioInference(api_client=self.index_api.api_client)
165        return self._inference

Dynamically create and cache the Inference instance.

async 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: 
167    async def create_index(
168        self,
169        name: str,
170        spec: Union[Dict, ServerlessSpec, PodSpec],
171        dimension: Optional[int] = None,
172        metric: Optional[Union[Metric, str]] = Metric.COSINE,
173        timeout: Optional[int] = None,
174        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
175        vector_type: Optional[Union[VectorType, str]] = VectorType.DENSE,
176        tags: Optional[Dict[str, str]] = None,
177    ) -> IndexModel:
178        req = PineconeDBControlRequestFactory.create_index_request(
179            name=name,
180            spec=spec,
181            dimension=dimension,
182            metric=metric,
183            deletion_protection=deletion_protection,
184            vector_type=vector_type,
185            tags=tags,
186        )
187        resp = await self.index_api.create_index(create_index_request=req)
188
189        if timeout == -1:
190            return IndexModel(resp)
191        return await 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
import asyncio

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

async def main():
    async with PineconeAsyncio(api_key=os.environ.get("PINECONE_API_KEY")) as pc:
        await 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"
            }
        )

asyncio.run(main())

Creating a pod index

import os
import asyncio

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

async def main():
    async with Pinecone(api_key=os.environ.get("PINECONE_API_KEY")) as pc:
        await 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"
            }
        )

asyncio.run(main())
async 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: 
193    async def create_index_for_model(
194        self,
195        name: str,
196        cloud: Union[CloudProvider, str],
197        region: Union[AwsRegion, GcpRegion, AzureRegion, str],
198        embed: Union[IndexEmbed, CreateIndexForModelEmbedTypedDict],
199        tags: Optional[Dict[str, str]] = None,
200        deletion_protection: Optional[Union[DeletionProtection, str]] = DeletionProtection.DISABLED,
201        timeout: Optional[int] = None,
202    ) -> IndexModel:
203        req = PineconeDBControlRequestFactory.create_index_for_model_request(
204            name=name,
205            cloud=cloud,
206            region=region,
207            embed=embed,
208            tags=tags,
209            deletion_protection=deletion_protection,
210        )
211        resp = await self.index_api.create_index_for_model(req)
212
213        if timeout == -1:
214            return IndexModel(resp)
215        return await 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.

import asyncio

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

async def main():
    async with PineconeAsyncio() as pc:
        if not await 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",
                    },
                )
            )

asyncio.run(main())

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.

async def delete_index(self, name: str, timeout: Optional[int] = None): 
255    async def delete_index(self, name: str, timeout: Optional[int] = None):
256        await self.index_api.delete_index(name)
257
258        if timeout == -1:
259            return
260
261        if timeout is None:
262            while await self.has_index(name):
263                await asyncio.sleep(5)
264        else:
265            while await self.has_index(name) and timeout >= 0:
266                await asyncio.sleep(5)
267                timeout -= 5
268        if timeout and timeout < 0:
269            raise (
270                TimeoutError(
271                    "Please call the list_indexes API ({}) to confirm if index is deleted".format(
272                        "https://www.pinecone.io/docs/api/operation/list_indexes/"
273                    )
274                )
275            )
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.

import asyncio

from pinecone import PineconeAsyncio

async def main():
    pc = PineconeAsyncio()

    index_name = "my_index"
    desc = await 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.
        await pc.configure_index(name=index_name, deletion_protection="disabled")

    await pc.delete_index(name=index_name)
    await pc.close()

asyncio.run(main())
async def list_indexes(self) -> pinecone.models.index_list.IndexList: 
277    async def list_indexes(self) -> IndexList:
278        response = await self.index_api.list_indexes()
279        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.

import asyncio

from pinecone import PineconeAsyncio

async def main():
    pc = PineconeAsyncio()

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

    await pc.close()

asyncio.run(main())
async def describe_index(self, name: str) -> pinecone.models.index_model.IndexModel: 
281    async def describe_index(self, name: str) -> IndexModel:
282        description = await self.index_api.describe_index(name)
283        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.

import asyncio
from pinecone import Pinecone, PineconeAsyncio, Index

async def main():
    pc = PineconeAsyncio()

    index_name="my_index"
    description = await 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}")
    await pc.close()

    async with Pinecone().IndexAsyncio(host=description.host) as idx:
        await idx.upsert(vectors=[...])

asyncio.run(main())
async def has_index(self, name: str) -> bool: 
285    async def has_index(self, name: str) -> bool:
286        available_indexes = await self.list_indexes()
287        if name in available_indexes.names():
288            return True
289        else:
290            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.

import asyncio
from pinecone import PineconeAsyncio, ServerlessSpec

async def main():
    async with PineconeAsyncio() as pc:
        index_name = "my_index"
        if not await 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")
            )

asyncio.run(main())
async 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): 
292    async def configure_index(
293        self,
294        name: str,
295        replicas: Optional[int] = None,
296        pod_type: Optional[Union[PodType, str]] = None,
297        deletion_protection: Optional[Union[DeletionProtection, str]] = None,
298        tags: Optional[Dict[str, str]] = None,
299    ):
300        description = await self.describe_index(name=name)
301
302        req = PineconeDBControlRequestFactory.configure_index_request(
303            description=description,
304            replicas=replicas,
305            pod_type=pod_type,
306            deletion_protection=deletion_protection,
307            tags=tags,
308        )
309        await self.index_api.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.

import asyncio
from pinecone import PineconeAsyncio, PodType

async def main():
    async with PineconeAsyncio() as pc:
        await pc.configure_index(
            name="my_index",
            replicas=2,
            pod_type=PodType.P1_X2
        )

asyncio.run(main())

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.

import asyncio
from pinecone import PineconeAsyncio, DeletionProtection

async def main():
    async with PineconeAsyncio() as pc:
        # Enable deletion protection
        await pc.configure_index(
            name="my_index",
            deletion_protection=DeletionProtection.ENABLED
        )

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

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

asyncio.run(main())

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.

import asyncio

from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio() as pc:
        # Add a tag
        await pc.configure_index(name="my_index", tags={"environment": "staging"})

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

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

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

asyncio.run(main())
async def create_collection(self, name: str, source: str): 
311    async def create_collection(self, name: str, source: str):
312        req = PineconeDBControlRequestFactory.create_collection_request(name=name, source=source)
313        await 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
async def list_collections(self) -> pinecone.models.collection_list.CollectionList: 
315    async def list_collections(self) -> CollectionList:
316        response = await self.index_api.list_collections()
317        return CollectionList(response)

List all collections

import asyncio
from pinecone import PineconeAsyncio

async def main():
    pc = PineconeAsyncio()

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

    # You can also iterate specifically over
    # a list of collection names by calling
    # the .names() helper.
    collection_name = "my_collection"
    collections = await pc.list_collections()
    if collection_name in collections.names():
        print('Collection exists')

    await pc.close()

asyncio.run(main())
async def delete_collection(self, name: str): 
319    async def delete_collection(self, name: str):
320        await self.index_api.delete_collection(name)

Describes a collection.

Parameters
  • The name of the collection
Returns

Description of the collection

import asyncio
from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio() as pc:

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

asyncio.run(main())
async def describe_collection(self, name: str): 
322    async def describe_collection(self, name: str):
323        return await self.index_api.describe_collection(name).to_dict()

Describes a collection.

Parameters
  • The name of the collection
Returns

Description of the collection

import asyncio
from pinecone import PineconeAsyncio


async def main():
    async with PineconeAsyncio() as pc:
        description = await pc.describe_collection("my_collection")
        print(description.name)
        print(description.source)
        print(description.status)
        print(description.size)

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

Build an asyncio-compatible client for index data operations.

Parameters
  • host: The host url of the index.
import os
import asyncio

from pinecone import PineconeAsyncio

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

async def main():
    async with Pinecone(api_key=api_key) as pc:
        async with pc.Index(host=index_host) as idx:
            # Now you're ready to perform data operations
            await index.query(vector=[...], top_k=10)

asyncio.run(main())

To find your host url, you can use the describe_index. Or, alternatively, the host is displayed in the Pinecone web console.

import os
import asyncio

from pinecone import PineconeAsyncio

async def main():
    async with PineconeAsyncio(
        api_key=os.environ.get("PINECONE_API_KEY")
    ) as pc:
        host = await pc.describe_index('index-name').host

asyncio.run(main())

Alternative setup

Like instances of the PineconeAsyncio class, instances of IndexAsyncio have async context that needs to be cleaned up when you are done with it in order to avoid error messages about unclosed session from aiohttp. Nesting these in code is a bit cumbersome, so if you are only planning to do data operations you may prefer to setup the IndexAsyncio object via the Pinecone class which will avoid creating an outer async context.

import os
import asyncio
from pinecone import Pinecone

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

async def main():
    pc = Pinecone(api_key=api_key) # sync client, so no async context to worry about

    async with pc.AsyncioIndex(host='your_index_host') as idx:
        # Now you're ready to perform data operations
        await idx.query(vector=[...], top_k=10)