Provisioning

Source code in superme_sdk/services/_provision.py

class ProvisionMixin:
    def provision_create(
        self,
        community_id: str,
        *,
        name: str,
        linkedin_url: str,
        contact_email: Optional[str] = None,
        notes: Optional[str] = None,
        socials: Optional[dict[str, str]] = None,
        external_urls: Optional[list[str]] = None,
    ) -> ProvisionCreateResponse:
        """Provision a single community member.

        Example:
            ```python
            result = client.provision_create(
                "community_abc",
                name="Jane Smith",
                linkedin_url="https://linkedin.com/in/janesmith",
                contact_email="jane@example.com",
                notes="Met at Dubai summit",
            )
            print(result["provision"]["user_id"])
            ```

        Args:
            community_id: The community to provision into.
            name: Full name of the person being provisioned.
            linkedin_url: LinkedIn profile URL (``linkedin.com/in/<slug>``).
            contact_email: Email address for invite delivery.
            notes: Community-scoped note (e.g. intro context, WhatsApp note).
            socials: Platform handles, e.g. ``{"x": "janesmith"}``.
            external_urls: URLs to import into the person's knowledge base.

        Returns:
            :class:`~superme_sdk.ProvisionCreateResponse` with the provisioned member record.
        """
        body = _provision_body(
            community_id,
            name,
            linkedin_url,
            contact_email,
            notes,
            socials,
            external_urls,
        )
        resp = self._rest_http.post(f"/api/v3/provision/{community_id}", json=body)
        self._check_rest_response(resp)
        return resp.json()

    def provision_create_batch(
        self,
        community_id: str,
        profiles: list[ProvisionProfile],
    ) -> list[dict[str, Any]]:
        """Provision multiple community members concurrently.

        Fans out up to 10 concurrent requests using a dedicated batch-scoped
        HTTP client (avoids sharing the main client across threads). Results
        are returned in the same order as ``profiles``. Failed items have an
        ``"error"`` key instead of ``"provision"``.

        Example:
            ```python
            results = client.provision_create_batch(
                "community_abc",
                profiles=[
                    {
                        "name": "Jane Smith",
                        "linkedin_url": "https://linkedin.com/in/janesmith",
                        "notes": "WhatsApp intro: loves B2B SaaS",
                    },
                    {
                        "name": "John Doe",
                        "linkedin_url": "https://linkedin.com/in/johndoe",
                        "contact_email": "john@example.com",
                    },
                ],
            )
            for r in results:
                if "error" in r:
                    print("failed:", r["error"])
                else:
                    print("provisioned:", r["provision"]["user_id"])
            ```

        Args:
            community_id: The community to provision into.
            profiles: List of profile dicts. Each supports the same keys as
                :meth:`provision_create` (``name``, ``linkedin_url``,
                ``contact_email``, ``notes``, ``socials``, ``external_urls``).

        Returns:
            List of result dicts in the same order as ``profiles``.
        """
        results: list[dict[str, Any]] = [{}] * len(profiles)

        # Fresh client scoped to this batch — avoids sharing self._rest_http across threads.
        batch_client = httpx.Client(
            base_url=self.rest_base_url,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
                "Accept": "application/json, text/event-stream",
            },
            timeout=self._rest_http.timeout,
        )

        def _one(index: int, profile: dict[str, Any]) -> None:
            try:
                body = _provision_body(community_id, **profile)
                resp = batch_client.post(f"/api/v3/provision/{community_id}", json=body)
                self._check_rest_response(resp)
                results[index] = resp.json()
            except Exception as exc:  # noqa: BLE001
                results[index] = {"error": str(exc)}

        try:
            with concurrent.futures.ThreadPoolExecutor(
                max_workers=_BATCH_CONCURRENCY
            ) as pool:
                futs = [pool.submit(_one, i, p) for i, p in enumerate(profiles)]
                concurrent.futures.wait(futs)
        finally:
            batch_client.close()

        return results

    def provision_send_invites(
        self,
        community_id: str,
        user_ids: list[str],
    ) -> ProvisionInviteResponse:
        """Send invite emails to provisioned members.

        Example:
            ```python
            result = client.provision_send_invites(
                "community_abc",
                user_ids=["user_123", "user_456"],
            )
            print(result["sent"], result["skipped"], result["failed"])
            ```

        Args:
            community_id: The community whose provisions to invite.
            user_ids: List of provisioned user IDs to send invites to.

        Returns:
            :class:`~superme_sdk.ProvisionInviteResponse` with ``sent``, ``skipped``, and ``failed`` lists.
        """
        resp = self._rest_http.post(
            f"/api/v3/provision/{community_id}/invite",
            json={"community_id": community_id, "user_ids": user_ids},
        )
        self._check_rest_response(resp)
        return resp.json()

    def provision_list(self, community_id: str) -> ProvisionListResponse:
        """List all provisions for a community.

        Example:
            ```python
            result = client.provision_list("community_abc")
            for p in result["provisions"]:
                print(p["user_id"], p["status"])
            ```

        Args:
            community_id: The community to list provisions for.

        Returns:
            :class:`~superme_sdk.ProvisionListResponse` with ``provisions`` list and ``count``.
        """
        resp = self._rest_http.get(
            f"/api/v3/provision/{community_id}",
            params={"community_id": community_id},
        )
        self._check_rest_response(resp)
        return resp.json()

    def provision_get(self, community_id: str, user_id: str) -> ProvisionRecord:
        """Fetch a single provisioned member by user ID.

        Example:
            ```python
            record = client.provision_get("community_abc", "user_123")
            print(record["status"], record["claim_url"])
            ```

        Args:
            community_id: The community that owns the provision.
            user_id: The provisioned user's ID.

        Returns:
            :class:`~superme_sdk.ProvisionRecord` for the given user.
        """
        resp = self._rest_http.get(f"/api/v3/provision/{community_id}/{user_id}")
        self._check_rest_response(resp)
        return resp.json()["provision"]

provision_create

provision_create(
    community_id: str,
    *,
    name: str,
    linkedin_url: str,
    contact_email: Optional[str] = None,
    notes: Optional[str] = None,
    socials: Optional[dict[str, str]] = None,
    external_urls: Optional[list[str]] = None,
) -> ProvisionCreateResponse

Provision a single community member.

Example

result = client.provision_create(
    "community_abc",
    name="Jane Smith",
    linkedin_url="https://linkedin.com/in/janesmith",
    contact_email="jane@example.com",
    notes="Met at Dubai summit",
)
print(result["provision"]["user_id"])

Parameters:

Name	Type	Description	Default
community_id	str	The community to provision into.	required
name	str	Full name of the person being provisioned.	required
linkedin_url	str	LinkedIn profile URL (linkedin.com/in/<slug>).	required
contact_email	Optional[str]	Email address for invite delivery.	None
notes	Optional[str]	Community-scoped note (e.g. intro context, WhatsApp note).	None
socials	Optional[dict[str, str]]	Platform handles, e.g. {"x": "janesmith"}.	None
external_urls	Optional[list[str]]	URLs to import into the person's knowledge base.	None

Returns:

Type	Description
ProvisionCreateResponse	class:~superme_sdk.ProvisionCreateResponse with the provisioned member record.

Source code in superme_sdk/services/_provision.py

def provision_create(
    self,
    community_id: str,
    *,
    name: str,
    linkedin_url: str,
    contact_email: Optional[str] = None,
    notes: Optional[str] = None,
    socials: Optional[dict[str, str]] = None,
    external_urls: Optional[list[str]] = None,
) -> ProvisionCreateResponse:
    """Provision a single community member.

    Example:
        ```python
        result = client.provision_create(
            "community_abc",
            name="Jane Smith",
            linkedin_url="https://linkedin.com/in/janesmith",
            contact_email="jane@example.com",
            notes="Met at Dubai summit",
        )
        print(result["provision"]["user_id"])
        ```

    Args:
        community_id: The community to provision into.
        name: Full name of the person being provisioned.
        linkedin_url: LinkedIn profile URL (``linkedin.com/in/<slug>``).
        contact_email: Email address for invite delivery.
        notes: Community-scoped note (e.g. intro context, WhatsApp note).
        socials: Platform handles, e.g. ``{"x": "janesmith"}``.
        external_urls: URLs to import into the person's knowledge base.

    Returns:
        :class:`~superme_sdk.ProvisionCreateResponse` with the provisioned member record.
    """
    body = _provision_body(
        community_id,
        name,
        linkedin_url,
        contact_email,
        notes,
        socials,
        external_urls,
    )
    resp = self._rest_http.post(f"/api/v3/provision/{community_id}", json=body)
    self._check_rest_response(resp)
    return resp.json()

provision_create_batch

provision_create_batch(
    community_id: str, profiles: list[ProvisionProfile]
) -> list[dict[str, Any]]

Provision multiple community members concurrently.

Fans out up to 10 concurrent requests using a dedicated batch-scoped HTTP client (avoids sharing the main client across threads). Results are returned in the same order as profiles. Failed items have an "error" key instead of "provision".

Example

results = client.provision_create_batch(
    "community_abc",
    profiles=[
        {
            "name": "Jane Smith",
            "linkedin_url": "https://linkedin.com/in/janesmith",
            "notes": "WhatsApp intro: loves B2B SaaS",
        },
        {
            "name": "John Doe",
            "linkedin_url": "https://linkedin.com/in/johndoe",
            "contact_email": "john@example.com",
        },
    ],
)
for r in results:
    if "error" in r:
        print("failed:", r["error"])
    else:
        print("provisioned:", r["provision"]["user_id"])

Parameters:

Name	Type	Description	Default
community_id	str	The community to provision into.	required
profiles	list[ProvisionProfile]	List of profile dicts. Each supports the same keys as :meth:provision_create (name, linkedin_url, contact_email, notes, socials, external_urls).	required

Returns:

Type	Description
list[dict[str, Any]]	List of result dicts in the same order as profiles.

Source code in superme_sdk/services/_provision.py

def provision_create_batch(
    self,
    community_id: str,
    profiles: list[ProvisionProfile],
) -> list[dict[str, Any]]:
    """Provision multiple community members concurrently.

    Fans out up to 10 concurrent requests using a dedicated batch-scoped
    HTTP client (avoids sharing the main client across threads). Results
    are returned in the same order as ``profiles``. Failed items have an
    ``"error"`` key instead of ``"provision"``.

    Example:
        ```python
        results = client.provision_create_batch(
            "community_abc",
            profiles=[
                {
                    "name": "Jane Smith",
                    "linkedin_url": "https://linkedin.com/in/janesmith",
                    "notes": "WhatsApp intro: loves B2B SaaS",
                },
                {
                    "name": "John Doe",
                    "linkedin_url": "https://linkedin.com/in/johndoe",
                    "contact_email": "john@example.com",
                },
            ],
        )
        for r in results:
            if "error" in r:
                print("failed:", r["error"])
            else:
                print("provisioned:", r["provision"]["user_id"])
        ```

    Args:
        community_id: The community to provision into.
        profiles: List of profile dicts. Each supports the same keys as
            :meth:`provision_create` (``name``, ``linkedin_url``,
            ``contact_email``, ``notes``, ``socials``, ``external_urls``).

    Returns:
        List of result dicts in the same order as ``profiles``.
    """
    results: list[dict[str, Any]] = [{}] * len(profiles)

    # Fresh client scoped to this batch — avoids sharing self._rest_http across threads.
    batch_client = httpx.Client(
        base_url=self.rest_base_url,
        headers={
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
        },
        timeout=self._rest_http.timeout,
    )

    def _one(index: int, profile: dict[str, Any]) -> None:
        try:
            body = _provision_body(community_id, **profile)
            resp = batch_client.post(f"/api/v3/provision/{community_id}", json=body)
            self._check_rest_response(resp)
            results[index] = resp.json()
        except Exception as exc:  # noqa: BLE001
            results[index] = {"error": str(exc)}

    try:
        with concurrent.futures.ThreadPoolExecutor(
            max_workers=_BATCH_CONCURRENCY
        ) as pool:
            futs = [pool.submit(_one, i, p) for i, p in enumerate(profiles)]
            concurrent.futures.wait(futs)
    finally:
        batch_client.close()

    return results

provision_send_invites

provision_send_invites(
    community_id: str, user_ids: list[str]
) -> ProvisionInviteResponse

Send invite emails to provisioned members.

Example

result = client.provision_send_invites(
    "community_abc",
    user_ids=["user_123", "user_456"],
)
print(result["sent"], result["skipped"], result["failed"])

Parameters:

Name	Type	Description	Default
community_id	str	The community whose provisions to invite.	required
user_ids	list[str]	List of provisioned user IDs to send invites to.	required

Returns:

Type	Description
ProvisionInviteResponse	class:~superme_sdk.ProvisionInviteResponse with sent, skipped, and failed lists.

Source code in superme_sdk/services/_provision.py

def provision_send_invites(
    self,
    community_id: str,
    user_ids: list[str],
) -> ProvisionInviteResponse:
    """Send invite emails to provisioned members.

    Example:
        ```python
        result = client.provision_send_invites(
            "community_abc",
            user_ids=["user_123", "user_456"],
        )
        print(result["sent"], result["skipped"], result["failed"])
        ```

    Args:
        community_id: The community whose provisions to invite.
        user_ids: List of provisioned user IDs to send invites to.

    Returns:
        :class:`~superme_sdk.ProvisionInviteResponse` with ``sent``, ``skipped``, and ``failed`` lists.
    """
    resp = self._rest_http.post(
        f"/api/v3/provision/{community_id}/invite",
        json={"community_id": community_id, "user_ids": user_ids},
    )
    self._check_rest_response(resp)
    return resp.json()

provision_list

provision_list(community_id: str) -> ProvisionListResponse

List all provisions for a community.

Example

result = client.provision_list("community_abc")
for p in result["provisions"]:
    print(p["user_id"], p["status"])

Parameters:

Name	Type	Description	Default
community_id	str	The community to list provisions for.	required

Returns:

Type	Description
ProvisionListResponse	class:~superme_sdk.ProvisionListResponse with provisions list and count.

Source code in superme_sdk/services/_provision.py

def provision_list(self, community_id: str) -> ProvisionListResponse:
    """List all provisions for a community.

    Example:
        ```python
        result = client.provision_list("community_abc")
        for p in result["provisions"]:
            print(p["user_id"], p["status"])
        ```

    Args:
        community_id: The community to list provisions for.

    Returns:
        :class:`~superme_sdk.ProvisionListResponse` with ``provisions`` list and ``count``.
    """
    resp = self._rest_http.get(
        f"/api/v3/provision/{community_id}",
        params={"community_id": community_id},
    )
    self._check_rest_response(resp)
    return resp.json()

provision_get

provision_get(
    community_id: str, user_id: str
) -> ProvisionRecord

Fetch a single provisioned member by user ID.

Example

record = client.provision_get("community_abc", "user_123")
print(record["status"], record["claim_url"])

Parameters:

Name	Type	Description	Default
community_id	str	The community that owns the provision.	required
user_id	str	The provisioned user's ID.	required

Returns:

Type	Description
ProvisionRecord	class:~superme_sdk.ProvisionRecord for the given user.

Source code in superme_sdk/services/_provision.py

def provision_get(self, community_id: str, user_id: str) -> ProvisionRecord:
    """Fetch a single provisioned member by user ID.

    Example:
        ```python
        record = client.provision_get("community_abc", "user_123")
        print(record["status"], record["claim_url"])
        ```

    Args:
        community_id: The community that owns the provision.
        user_id: The provisioned user's ID.

    Returns:
        :class:`~superme_sdk.ProvisionRecord` for the given user.
    """
    resp = self._rest_http.get(f"/api/v3/provision/{community_id}/{user_id}")
    self._check_rest_response(resp)
    return resp.json()["provision"]