from typing import Optional
from pinecone.exceptions import NotFoundException, PineconeException
from pinecone.openapi_support import ApiClient
from pinecone.core.openapi.admin.apis import ProjectsApi
from pinecone.utils import parse_non_empty_args, require_kwargs
from pinecone.core.openapi.admin.models import CreateProjectRequest, UpdateProjectRequest
import logging
import time
logger = logging.getLogger(__name__)
[docs]
class ProjectResource:
"""
This class is used to create, delete, list, fetch, and update projects.
.. note::
The class should not be instantiated directly. Instead, access this classes
methods through the :class:`pinecone.Admin` class's
:attr:`project` or :attr:`projects` attributes.
.. code-block:: python
from pinecone import Admin
# Credentials read from PINECONE_CLIENT_ID and
# PINECONE_CLIENT_SECRET environment variables
admin = Admin()
# Now call project methods on the projects namespace
project = admin.projects.create(
name="my-project",
max_pods=10,
force_encryption_with_cmek=False
)
"""
def __init__(self, api_client: ApiClient):
"""
Initialize the ProjectResource.
.. warning::
This class should not be instantiated directly. Instead, access this classes
methods through the :class:`pinecone.Admin` class's
:attr:`project` or :attr:`projects` attributes.
:param api_client: The API client to use.
:type api_client: ApiClient
"""
self._projects_api = ProjectsApi(api_client=api_client)
self._api_client = api_client
[docs]
@require_kwargs
def list(self):
"""
List all projects in the organization.
:return: An object with a list of projects.
:rtype: {"data": [Project]}
.. code-block:: python
:caption: List all projects in the organization
:emphasize-lines: 8
from pinecone import Admin
# Credentials read from PINECONE_CLIENT_ID and
# PINECONE_CLIENT_SECRET environment variables
admin = Admin()
# List all projects in the organization
projects_response = admin.projects.list()
for project in projects_response.data:
print(project.id)
print(project.name)
print(project.max_pods)
print(project.force_encryption_with_cmek)
"""
return self._projects_api.list_projects()
[docs]
@require_kwargs
def fetch(self, project_id: Optional[str] = None, name: Optional[str] = None):
"""
Fetch a project by project_id or name.
:param project_id: The project_id of the project to fetch.
:type project_id: str
:param name: The name of the project to fetch.
:type name: str
:return: The project.
:rtype: Project
Examples
--------
.. code-block:: python
:caption: Fetch a project by project_id
:emphasize-lines: 7-9
from pinecone import Admin
# Credentials read from PINECONE_CLIENT_ID and
# PINECONE_CLIENT_SECRET environment variables
admin = Admin()
project = admin.projects.fetch(
project_id="42ca341d-43bf-47cb-9f27-e645dbfabea6"
)
print(project.id)
print(project.name)
print(project.max_pods)
print(project.force_encryption_with_cmek)
print(project.organization_id)
print(project.created_at)
.. code-block:: python
:caption: Fetch a project by name
:emphasize-lines: 7
from pinecone import Admin
# Credentials read from PINECONE_CLIENT_ID and
# PINECONE_CLIENT_SECRET environment variables
admin = Admin()
project = admin.projects.fetch(name="my-project-name")
print(project.id)
print(project.name)
print(project.max_pods)
print(project.force_encryption_with_cmek)
print(project.organization_id)
print(project.created_at)
"""
if project_id is not None and name is not None:
raise ValueError("Either project_id or name must be provided but not both")
elif project_id is None and name is None:
raise ValueError("Either project_id or name must be provided")
if project_id is not None:
return self._projects_api.fetch_project(project_id=project_id)
else:
projects = self.list().data
projects = [project for project in projects if project.name == name]
if len(projects) == 0:
raise NotFoundException(f"Project with name '{name}' not found")
elif len(projects) > 1:
ids = [project.id for project in projects]
raise PineconeException(
f"Multiple projects found with name '{name}'. Please use project_id to fetch a specific project. Matching project ids: {ids}"
)
else:
return projects[0]
[docs]
@require_kwargs
def get(self, project_id: Optional[str] = None, name: Optional[str] = None):
"""Alias for :func:`fetch`
Examples
--------
.. code-block:: python
:caption: Get a project by project_id
:emphasize-lines: 7-9
from pinecone import Admin
# Credentials read from PINECONE_CLIENT_ID and
# PINECONE_CLIENT_SECRET environment variables
admin = Admin()
project = admin.project.get(
project_id="42ca341d-43bf-47cb-9f27-e645dbfabea6"
)
print(project.id)
print(project.name)
print(project.max_pods)
print(project.force_encryption_with_cmek)
"""
return self.fetch(project_id=project_id, name=name)
[docs]
@require_kwargs
def describe(self, project_id: Optional[str] = None, name: Optional[str] = None):
"""Alias for :func:`fetch`
Examples
--------
.. code-block:: python
:caption: Describe a project by project_id
:emphasize-lines: 7-9
from pinecone import Admin
# Credentials read from PINECONE_CLIENT_ID and
# PINECONE_CLIENT_SECRET environment variables
admin = Admin()
project = admin.project.describe(
project_id="42ca341d-43bf-47cb-9f27-e645dbfabea6"
)
print(project.id)
print(project.name)
print(project.max_pods)
print(project.force_encryption_with_cmek)
"""
return self.fetch(project_id=project_id, name=name)
[docs]
@require_kwargs
def exists(self, project_id: Optional[str] = None, name: Optional[str] = None):
"""
Check if a project exists by project_id or name.
:param project_id: The project_id of the project to check.
:type project_id: str
:param name: The name of the project to check.
:type name: str
:return: True if the project exists, False otherwise.
:rtype: bool
:raises ValueError: If both project_id and name are provided.
Examples
--------
.. code-block:: python
:caption: Check if a project exists by project name
:emphasize-lines: 8
from pinecone import Admin
# Credentials read from PINECONE_CLIENT_ID and
# PINECONE_CLIENT_SECRET environment variables
admin = Admin()
project_name = "my-project-name"
if admin.project.exists(name=project_name):
print(f"Project {project_name} exists")
else:
admin.project.create(
name=project_name,
max_pods=10,
force_encryption_with_cmek=False
)
.. code-block:: python
:caption: Check if a project exists by project_id
:emphasize-lines: 8
from pinecone import Admin
# Credentials read from PINECONE_CLIENT_ID and
# PINECONE_CLIENT_SECRET environment variables
admin = Admin()
project_id = "42ca341d-43bf-47cb-9f27-e645dbfabea6"
if admin.project.exists(project_id=project_id):
print(f"Project {project_id} exists")
else:
print(f"Project {project_id} does not exist")
"""
if project_id is not None and name is not None:
raise ValueError("Either project_id or name must be provided but not both")
elif project_id is None and name is None:
raise ValueError("Either project_id or name must be provided")
try:
args = [("project_id", project_id), ("name", name)]
self.fetch(**parse_non_empty_args(args))
return True
except NotFoundException:
return False
[docs]
@require_kwargs
def create(
self,
name: str,
max_pods: Optional[int] = None,
force_encryption_with_cmek: Optional[bool] = None,
):
"""
Create a project.
:param name: The name of the project to create.
:type name: str
:param max_pods: The maximum number of pods for the project.
:type max_pods: int
:param force_encryption_with_cmek: Whether to force encryption with CMEK.
:type force_encryption_with_cmek: bool
:return: The created project.
:rtype: Project
Examples
--------
.. code-block:: python
:caption: Create a project
:emphasize-lines: 7-11
from pinecone import Admin
# Credentials read from PINECONE_CLIENT_ID and
# PINECONE_CLIENT_SECRET environment variables
admin = Admin()
project = admin.project.create(
name="my-project-name",
max_pods=10,
force_encryption_with_cmek=False
)
print(project.id)
print(project.name)
print(project.organization_id)
print(project.max_pods)
print(project.force_encryption_with_cmek)
print(project.created_at)
"""
args = [
("name", name),
("max_pods", max_pods),
("force_encryption_with_cmek", force_encryption_with_cmek),
]
create_request = CreateProjectRequest(**parse_non_empty_args(args))
return self._projects_api.create_project(create_project_request=create_request)
[docs]
@require_kwargs
def update(
self,
project_id: str,
name: Optional[str] = None,
max_pods: Optional[int] = None,
force_encryption_with_cmek: Optional[bool] = None,
):
"""
Update a project.
:param project_id: The project_id of the project to update.
:type project_id: str
:param name: The name of the project to update.
:type name: str
:param max_pods: The maximum number of pods for the project.
:type max_pods: int
:param force_encryption_with_cmek: Whether to force encryption with CMEK.
:type force_encryption_with_cmek: bool
:return: The updated project.
:rtype: Project
Examples
--------
.. code-block:: python
:caption: Update a project by project_id
:emphasize-lines: 10-13, 16-19
from pinecone import Admin
# Credentials read from PINECONE_CLIENT_ID and
# PINECONE_CLIENT_SECRET environment variables
admin = Admin()
project = admin.project.get(name='my-project-name')
# Update max pods to 10
project = admin.project.update(
project_id=project.id,
max_pods=10
)
# Update force_encryption_with_cmek to True
project = admin.project.update(
project_id=project.id,
force_encryption_with_cmek=True
)
"""
args = [
("name", name),
("max_pods", max_pods),
("force_encryption_with_cmek", force_encryption_with_cmek),
]
update_request = UpdateProjectRequest(**parse_non_empty_args(args))
return self._projects_api.update_project(
project_id=project_id, update_project_request=update_request
)
[docs]
@require_kwargs
def delete(
self,
project_id: str,
delete_all_indexes: bool = False,
delete_all_collections: bool = False,
delete_all_backups: bool = False,
):
"""
.. warning::
Deleting a project is a permanent and irreversible operation.
Please be very sure you want to delete the project and everything
associated with it before calling this function.
Projects can only be deleted if they are empty. The delete operation
will fail if the project contains any resources such as indexes,
collections, or backups.
If you pass additional options such as ``delete_all_indexes=True``,
``delete_all_collections=True``, or ``delete_all_backups=True``, this function
will attempt to delete all of these resources before deleting the project itself.
**These deletions are permanent and cannot be undone.**
:param project_id: The project_id of the project to delete.
:type project_id: str
:param delete_all_indexes: Attempt to delete all indexes associated with the project.
:type delete_all_indexes: bool
:param delete_all_collections: Attempt to delete all collections associated with the project.
:type delete_all_collections: bool
:param delete_all_backups: Attempt to delete all backups associated with the project.
:type delete_all_backups: bool
:return: ``None``
Examples
--------
.. code-block:: python
:caption: Delete a project by project_id
:emphasize-lines: 9
from pinecone import Admin
# Credentials read from PINECONE_CLIENT_ID and
# PINECONE_CLIENT_SECRET environment variables
admin = Admin()
project = admin.project.get(name='my-project-name')
admin.project.delete(project_id=project.id)
.. code-block:: python
:caption: Delete a project that still contains indexes, collections, and backups
:emphasize-lines: 7-12
from pinecone import Admin
admin = Admin()
project = admin.project.get(name='my-project-name')
admin.project.delete(
project_id=project.id,
delete_all_indexes=True,
delete_all_collections=True,
delete_all_backups=True
)
if not admin.project.exists(project_id=project.id):
print("Project deleted successfully")
else:
print("Project deletion failed")
"""
project = self.get(project_id=project_id)
if not (delete_all_indexes or delete_all_collections or delete_all_backups):
return self._projects_api.delete_project(project_id=project_id)
from .api_key import ApiKeyResource
api_key_resource = ApiKeyResource(self._api_client)
logger.debug(f"Creating API key 'cleanup-key' for project {project.id}")
key_create_response = api_key_resource.create(
project_id=project.id, name="cleanup-key", roles=["ProjectEditor"]
)
api_key = key_create_response.value
try:
from ..eraser.project_eraser import _ProjectEraser
done = False
retries = 0
while not done and retries < 5:
project_eraser = _ProjectEraser(api_key=api_key)
if delete_all_collections:
project_eraser.delete_all_collections()
if delete_all_backups:
project_eraser.delete_all_backups()
if delete_all_indexes:
project_eraser.delete_all_indexes()
done = not project_eraser.retry_needed()
retries += 1
if not done:
logger.debug(
f"Retrying deletion of resources for project {project.id}. There were {len(project_eraser.undeleteable_resources)} undeleteable resources"
)
time.sleep(30)
finally:
logger.debug(f"Deleting API key 'cleanup-key' for project {project.id}")
api_key_resource.delete(api_key_id=key_create_response.key.id)
logger.info(f"Deleting project {project_id}")
return self._projects_api.delete_project(project_id=project_id)
