Skip to content

Provisioning (async)

Source code in superme_sdk/services/aio/_provision.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class AsyncProvisionMixin:
    async 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 (async).

        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: dict[str, Any] = {
            "community_id": community_id,
            "name": name,
            "linkedin_url": linkedin_url,
        }
        if contact_email is not None:
            body["contact_email"] = contact_email
        if notes is not None:
            body["notes"] = notes
        if socials is not None:
            body["socials"] = socials
        if external_urls is not None:
            body["external_urls"] = external_urls

        resp = await self._async_rest_http.post(
            f"/api/v3/provision/{community_id}", json=body
        )
        self._check_rest_response(resp)
        return resp.json()

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

        Fans out up to 10 concurrent :meth:`provision_create` calls.
        Results are in the same order as ``profiles``; failed items have an
        ``"error"`` key instead of ``"provision"``.

        Args:
            community_id: The community to provision into.
            profiles: List of profile dicts (``name``, ``linkedin_url``,
                ``contact_email``, ``notes``, ``socials``, ``external_urls``).

        Returns:
            List of result dicts in the same order as ``profiles``.
        """
        sem = asyncio.Semaphore(_BATCH_CONCURRENCY)

        async def _one(profile: dict[str, Any]) -> dict[str, Any]:
            async with sem:
                return await self.provision_create(community_id, **profile)

        tasks = [asyncio.create_task(_one(p)) for p in profiles]
        results = []
        for task in tasks:
            try:
                results.append(await task)
            except Exception as exc:  # noqa: BLE001
                results.append({"error": str(exc)})
        return results

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

        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 = await self._async_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()

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

        Args:
            community_id: The community to list provisions for.

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

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

        Example:
            ```python
            record = await 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 = await self._async_rest_http.get(
            f"/api/v3/provision/{community_id}/{user_id}"
        )
        self._check_rest_response(resp)
        return resp.json()["provision"]

provision_create async 

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 (async).

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/aio/_provision.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
async 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 (async).

    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: dict[str, Any] = {
        "community_id": community_id,
        "name": name,
        "linkedin_url": linkedin_url,
    }
    if contact_email is not None:
        body["contact_email"] = contact_email
    if notes is not None:
        body["notes"] = notes
    if socials is not None:
        body["socials"] = socials
    if external_urls is not None:
        body["external_urls"] = external_urls

    resp = await self._async_rest_http.post(
        f"/api/v3/provision/{community_id}", json=body
    )
    self._check_rest_response(resp)
    return resp.json()

provision_create_batch async 

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

Provision multiple community members concurrently (async).

Fans out up to 10 concurrent :meth:provision_create calls. Results are in the same order as profiles; failed items have an "error" key instead of "provision".

Parameters:

Name Type Description Default
community_id str

The community to provision into.

 required
profiles list[ProvisionProfile]

List of profile dicts (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/aio/_provision.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
async def provision_create_batch(
    self,
    community_id: str,
    profiles: list[ProvisionProfile],
) -> list[dict[str, Any]]:
    """Provision multiple community members concurrently (async).

    Fans out up to 10 concurrent :meth:`provision_create` calls.
    Results are in the same order as ``profiles``; failed items have an
    ``"error"`` key instead of ``"provision"``.

    Args:
        community_id: The community to provision into.
        profiles: List of profile dicts (``name``, ``linkedin_url``,
            ``contact_email``, ``notes``, ``socials``, ``external_urls``).

    Returns:
        List of result dicts in the same order as ``profiles``.
    """
    sem = asyncio.Semaphore(_BATCH_CONCURRENCY)

    async def _one(profile: dict[str, Any]) -> dict[str, Any]:
        async with sem:
            return await self.provision_create(community_id, **profile)

    tasks = [asyncio.create_task(_one(p)) for p in profiles]
    results = []
    for task in tasks:
        try:
            results.append(await task)
        except Exception as exc:  # noqa: BLE001
            results.append({"error": str(exc)})
    return results

provision_send_invites async 

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

Send invite emails to provisioned members (async).

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/aio/_provision.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
async def provision_send_invites(
    self,
    community_id: str,
    user_ids: list[str],
) -> ProvisionInviteResponse:
    """Send invite emails to provisioned members (async).

    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 = await self._async_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 async 

provision_list(community_id: str) -> ProvisionListResponse

List all provisions for a community (async).

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/aio/_provision.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def provision_list(self, community_id: str) -> ProvisionListResponse:
    """List all provisions for a community (async).

    Args:
        community_id: The community to list provisions for.

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

provision_get async 

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

Fetch a single provisioned member by user ID (async).

Example 
record = await 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/aio/_provision.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
async def provision_get(self, community_id: str, user_id: str) -> ProvisionRecord:
    """Fetch a single provisioned member by user ID (async).

    Example:
        ```python
        record = await 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 = await self._async_rest_http.get(
        f"/api/v3/provision/{community_id}/{user_id}"
    )
    self._check_rest_response(resp)
    return resp.json()["provision"]