"""Invites namespace: list, create, describe, delete, and resend organization invites."""
from __future__ import annotations
import logging
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any
from urllib.parse import quote
from pinecone._internal.adapters.admin_adapter import AdminAdapter
from pinecone._internal.role_bindings import normalize_role_bindings
from pinecone._internal.validation import require_in_range, require_non_empty
from pinecone.errors.exceptions import ValidationError
from pinecone.models.admin.invite import InviteModel
from pinecone.models.admin.role_binding import RoleBindingInput
from pinecone.models.pagination import Page, Paginator
if TYPE_CHECKING:
from pinecone._internal.http_client import HTTPClient
logger = logging.getLogger(__name__)
_LIMIT_MIN = 1
_LIMIT_MAX = 100
[docs]
class Invites:
"""Offers of organization membership that have not yet been accepted.
An invite is an emailed offer for someone to join the organization that the
:class:`~pinecone.Admin` client's OAuth credentials resolve to. It is a
principal in its own right (roles can be bound to it before anyone accepts),
and accepting it turns the recipient into a user. Not constructed directly;
reach it as ``admin.invites``.
An invite's role bindings are not part of its representation: :meth:`create`
sends them, but no method here returns them. Read or change them through
:class:`~pinecone.admin.role_bindings.RoleBindings` with
``principal_type="invite"``.
See
`Error handling <https://sdk.pinecone.io/python/guides/error-handling.html>`__
for the exceptions every operation here
can raise.
Examples:
>>> from pinecone import Admin
>>> admin = Admin(client_id="your-client-id", client_secret="your-client-secret")
>>> for invite in admin.invites.list():
... print(invite.email, invite.status)
newhire@acme.com pending
.. seealso::
- :class:`~pinecone.admin.users.Users`: the same person after they
accept. An invite and the user it produces are separate records with
separate IDs, and only one of the two appears in each list.
- :class:`~pinecone.admin.service_accounts.ServiceAccounts`: machine
identities, which are created directly and never invited.
"""
[docs]
def __init__(self, *, http: HTTPClient) -> None:
self._http = http
self._adapter = AdminAdapter()
def __repr__(self) -> str:
"""Return developer-friendly representation."""
return "Invites()"
[docs]
def list(
self,
*,
limit: int | None = None,
pagination_token: str | None = None,
) -> Paginator[InviteModel]:
"""List the organization's pending and expired invites, with lazy pagination.
Accepted invites are omitted, so absence from this list does not mean an
invite never existed (see the note below). No request is sent until the
returned paginator is iterated; see
`Pagination <https://sdk.pinecone.io/python/guides/pagination.html>`__.
Args:
limit (int | None): Number of invites the server returns **per
page**. It caps each page, not how many invites the paginator
yields in total; the paginator keeps following cursors until the
pages run out, so use :func:`itertools.islice` to cap the total.
When ``None`` the server chooses the page size.
pagination_token (str | None): Cursor from a previous paginator's
``pagination_token``, to resume where that iteration stopped.
Reuse it with the same ``limit``.
Returns:
:class:`~pinecone.models.pagination.Paginator` yielding
:class:`~pinecone.models.admin.invite.InviteModel` objects.
Raises:
:exc:`~pinecone.errors.exceptions.PineconeValueError`:
If *limit* is outside the allowed page-size range. Raised before any
network call.
Examples:
>>> from pinecone import Admin
>>> admin = Admin(client_id="your-client-id", client_secret="your-client-secret")
>>> for invite in admin.invites.list():
... print(invite.email, invite.status)
newhire@acme.com pending
Page-level access exposes the cursor, which is ``None`` once there
is no further page to fetch:
>>> for page in admin.invites.list(limit=25).pages():
... print(len(page.items), page.pagination_token)
1 None
.. note::
An invite missing from this list may simply have been accepted.
:meth:`describe` still returns it, with
``status == InviteStatus.PROCESSED``, and the accepted invitee is
now a member reachable through
:meth:`Users.list() <pinecone.admin.users.Users.list>`. To reconcile who has
access, read both lists.
.. seealso::
- :meth:`Users.list() <pinecone.admin.users.Users.list>`: the members this list's
invitees become once they accept.
"""
if limit is not None:
require_in_range("limit", limit, _LIMIT_MIN, _LIMIT_MAX)
logger.info("Listing invites (limit=%r)", limit)
def fetch_page(token: str | None) -> Page[InviteModel]:
params: dict[str, str | int] = {}
if limit is not None:
params["limit"] = limit
if token is not None:
params["paginationToken"] = token
response = self._http.get("/admin/invites", params=params)
result = self._adapter.to_invite_list(response.content)
logger.debug("Listed %d invites (has_more=%s)", len(result), result.has_more)
return Page(items=result.data, pagination_token=result.pagination_token)
return Paginator(fetch_page=fetch_page, initial_token=pagination_token)
[docs]
def create(
self,
*,
email: str,
role_bindings: Sequence[RoleBindingInput | Mapping[str, Any]],
) -> InviteModel:
"""Invite someone to the organization and grant their initial roles.
The server has already sent the email by the time this returns; the
invite comes back ``pending``, with ``expires_at`` set to when it lapses.
The response does **not** echo the role bindings.
Args:
email (str): The address to invite, e.g. ``"newhire@acme.com"``. The
SDK checks only that it isn't empty; the server validates the
address itself and rejects a malformed or over-long one.
role_bindings (Sequence[RoleBindingInput | Mapping[str, Any]]):
The roles to grant the invitee, as
:class:`~pinecone.models.admin.role_binding.RoleBindingInput`
instances or plain dicts, mixed freely. Each entry needs
``resource_type`` (``"organization"`` or ``"project"``) and
``role``; ``project`` scope additionally needs ``resource_id``,
the project UUID. At least one entry is required, and the server
requires at least one of them to be an ``organization``-scoped
membership role (``OrgOwner``, ``OrgManager``,
``OrgBillingAdmin``, or ``OrgMember``); the server rejects a
project-only invite.
Returns:
The created :class:`~pinecone.models.admin.invite.InviteModel`, whose
``id`` is what :meth:`describe`, :meth:`resend`, and :meth:`delete`
take, and what identifies the invite as a ``principal_id`` in
role-binding queries.
Raises:
:exc:`~pinecone.errors.exceptions.PineconeTypeError`: If *email* is not a ``str``.
:exc:`~pinecone.errors.exceptions.PineconeValueError`:
If *email* is empty, if *role_bindings* is empty, or if any
entry is missing ``resource_type``/``role``, carries an
unrecognized key, or names a value this SDK release does not
know. The message names the index of the offending entry.
Raised before any network call.
:exc:`~pinecone.errors.exceptions.ConflictError`:
If a pending invite already exists for the address, or the
address already belongs to a member. In the second case there is
nothing to invite; manage the existing user's roles instead.
Examples:
>>> from pinecone import Admin
>>> admin = Admin(client_id="your-client-id", client_secret="your-client-secret")
>>> invite = admin.invites.create(
... email="newhire@acme.com",
... role_bindings=[{"resource_type": "organization", "role": "OrgMember"}],
... )
>>> invite.status
'pending'
>>> invite.processed_at is None
True
The result is a pending principal, not a member: its own ``id`` is
what :meth:`resend`, :meth:`delete`, and role-binding queries take,
and the invitee stays absent from
:meth:`Users.list() <pinecone.admin.users.Users.list>` until they accept.
Typed inputs and dicts are interchangeable, and may be mixed:
>>> from pinecone.models.admin import ResourceType, RoleBindingInput, RoleName
>>> invite = admin.invites.create(
... email="newhire@acme.com",
... role_bindings=[
... RoleBindingInput(
... resource_type=ResourceType.ORGANIZATION,
... role=RoleName.ORG_MEMBER,
... ),
... {
... "resource_type": "project",
... "role": "ProjectViewer",
... "resource_id": "a2f7dddb-1597-4eff-9f71-535fde243f58",
... },
... ],
... )
>>> invite.email
'newhire@acme.com'
.. seealso::
- :meth:`RoleBindings.create()
<pinecone.admin.role_bindings.RoleBindings.create>`: how to grant
a further role after the invite exists, and the only way to read
back the roles this call sent.
- :meth:`ServiceAccounts.create()
<pinecone.admin.service_accounts.ServiceAccounts.create>`: the machine
equivalent, which takes the same binding shape but mints
credentials instead of emailing anyone.
"""
require_non_empty("email", email)
bindings = list(role_bindings)
if not bindings:
raise ValidationError(
"role_bindings must be a non-empty list; an invite needs at least one "
"role binding, including an 'organization'-scoped membership role"
)
body: dict[str, Any] = {"email": email, "role_bindings": normalize_role_bindings(bindings)}
logger.info("Creating invite (bindings=%d)", len(bindings))
response = self._http.post("/admin/invites", json=body)
result = self._adapter.to_invite(response.content)
logger.debug("Created invite %r", result.id)
return result
[docs]
def describe(self, *, invite_id: str) -> InviteModel:
"""Get one invite's details, whatever its status.
Unlike :meth:`list`, this reaches accepted invites too. It is the only
operation that can return ``status == InviteStatus.PROCESSED``, which is
how you tell an accepted invite from one that never existed.
Args:
invite_id (str): The invite's UUID, as carried by ``InviteModel.id``.
This is not the ID of the user the invite produced on
acceptance; the two records have separate IDs.
Returns:
An :class:`~pinecone.models.admin.invite.InviteModel`. On an accepted
invite, ``processed_at`` carries when it was accepted.
Raises:
:exc:`~pinecone.errors.exceptions.PineconeTypeError`: If *invite_id* is not a ``str``.
:exc:`~pinecone.errors.exceptions.PineconeValueError`: If *invite_id* is empty.
:exc:`~pinecone.errors.exceptions.NotFoundError`:
If no such invite exists in the organization. A deleted invite
reads back as not found rather than as a status value, so
not-found and accepted are genuinely different answers here.
Examples:
>>> from pinecone import Admin
>>> admin = Admin(client_id="your-client-id", client_secret="your-client-secret")
>>> invite = admin.invites.describe(
... invite_id="9c8e3528-b9c0-4358-84ce-84c28e91b566"
... )
>>> invite.email, invite.status
('newhire@acme.com', 'pending')
.. seealso::
- :meth:`Users.describe() <pinecone.admin.users.Users.describe>`: the member record
created when this invite was accepted, addressed by its own user ID.
"""
require_non_empty("invite_id", invite_id)
logger.info("Describing invite %r", invite_id)
response = self._http.get(f"/admin/invites/{quote(invite_id, safe='')}")
result = self._adapter.to_invite(response.content)
logger.debug("Described invite %r", invite_id)
return result
[docs]
def delete(self, *, invite_id: str) -> None:
"""Withdraw a pending or expired invite, along with its role bindings.
The invite and its role bindings are gone by the time this returns: a
repeat call, or fetching it by ID afterwards, gets a not-found error.
Args:
invite_id (str): The UUID of the invite to withdraw.
Raises:
:exc:`~pinecone.errors.exceptions.PineconeTypeError`: If *invite_id* is not a ``str``.
:exc:`~pinecone.errors.exceptions.PineconeValueError`: If *invite_id* is empty.
:exc:`~pinecone.errors.exceptions.NotFoundError`:
If no such invite exists in the organization.
:exc:`~pinecone.errors.exceptions.ConflictError`:
If the invite has already been accepted. There is no invite left
to withdraw; remove the resulting member with
:meth:`Users.delete() <pinecone.admin.users.Users.delete>` instead.
Examples:
>>> from pinecone import Admin
>>> admin = Admin(client_id="your-client-id", client_secret="your-client-secret")
>>> admin.invites.delete(invite_id="9c8e3528-b9c0-4358-84ce-84c28e91b566")
.. seealso::
- :meth:`Users.delete() <pinecone.admin.users.Users.delete>`: the only way to revoke
access once an invite has been accepted.
"""
require_non_empty("invite_id", invite_id)
logger.info("Deleting invite %r", invite_id)
self._http.delete(f"/admin/invites/{quote(invite_id, safe='')}")
logger.debug("Deleted invite %r", invite_id)
[docs]
def resend(self, *, invite_id: str) -> InviteModel:
"""Resend an invite's email and push its expiration back out.
Works on pending and expired invites alike: the returned invite is
``pending`` again with a fresh ``expires_at``.
Args:
invite_id (str): The UUID of the invite to resend.
Returns:
The updated :class:`~pinecone.models.admin.invite.InviteModel`, with
``status`` back to ``pending`` and a later ``expires_at``. Read the
new expiry from here rather than computing it.
Raises:
:exc:`~pinecone.errors.exceptions.PineconeTypeError`: If *invite_id* is not a ``str``.
:exc:`~pinecone.errors.exceptions.PineconeValueError`: If *invite_id* is empty.
:exc:`~pinecone.errors.exceptions.NotFoundError`:
If no such invite exists in the organization.
:exc:`~pinecone.errors.exceptions.ConflictError`:
If the invite has already been accepted. Do not retry; there is
nothing left to resend.
Examples:
>>> from pinecone import Admin
>>> admin = Admin(client_id="your-client-id", client_secret="your-client-secret")
>>> invite = admin.invites.resend(
... invite_id="9c8e3528-b9c0-4358-84ce-84c28e91b566"
... )
>>> invite.status
'pending'
>>> invite.expires_at
'2026-05-21T03:00:00Z'
"""
require_non_empty("invite_id", invite_id)
logger.info("Resending invite %r", invite_id)
response = self._http.post(f"/admin/invites/{quote(invite_id, safe='')}/resend")
result = self._adapter.to_invite(response.content)
logger.debug("Resent invite %r", invite_id)
return result