Skip to content

Sync Client

The Client is the synchronous entry-point to the Bitcaster API. Every method blocks on the HTTP request and returns the parsed response directly.

Quick Start

from bitcaster_sdk.client import Client

client = Client("https://key@app.bitcaster.io/api/o/ORG/")
client.set_domain("my-project", "my-app")
result = client.trigger_event("order-placed", context={"order_id": "123"})

API Reference

Bases: AbstractClient

Sync HTTP client for the Bitcaster REST API.

Every method blocks on the HTTP request and returns the parsed response directly.

Source code in src/bitcaster_sdk/client.py 
 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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
class Client(AbstractClient):
    """Sync HTTP client for the Bitcaster REST API.

    Every method blocks on the HTTP request and returns the parsed response
    directly.
    """

    _transport_class: type[AbstractTransport] = Transport

    def ping(self) -> dict[str, Any]:
        """Check connectivity with the Bitcaster server.

        Returns:
            A dict with server identifying information (e.g. ``token``, ``slug``).

        """
        try:
            response = self.transport.get("/api/system/ping/")
            self.assert_response(response)
            return response.json()
        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
            raise ConnectionError(f"Connection Error: {self.api_url}") from e
        except Exception as e:
            logger.exception(e)
            raise

    def list_events(self, project: str, application: str) -> list[dict[str, Any]]:
        """List events for a given project and application.

        Args:
            project: Project slug.
            application: Application slug.

        Returns:
            A list of event dicts, each containing ``name``, ``slug``,
            ``active``, ``locked``, ``description``.

        """
        try:
            response = self.transport.get(f"p/{project}/a/{application}/e/")
            self.assert_response(response)
            return response.json()
        except Exception as e:
            logger.exception(e)
            raise e

    def list_users(self) -> list[dict[str, Any]]:
        """List users in the current organization.

        Returns:
            A list of user dicts, each containing ``email``, ``username``,
            ``is_active``, ``locked``, etc.

        """
        try:
            response = self.transport.get("u/")
            self.assert_response(response)
            return response.json()
        except Exception as e:
            logger.exception(e)
            raise

    def list_distribution_lists(self, project: str) -> list[dict[str, Any]]:
        """List distribution lists for a project.

        Args:
            project: Project slug.

        Returns:
            A list of distribution list dicts.

        """
        try:
            response = self.transport.get(f"p/{project}/d/")
            self.assert_response(response)
            return response.json()
        except Exception as e:
            logger.exception(e)
            raise

    def list_projects(self) -> list[dict[str, Any]]:
        """List projects in the current organization.

        Returns:
            A list of project dicts, each containing ``slug``, ``name``,
            ``applications``, ``lists``, ``channels``.

        """
        try:
            response = self.transport.get("p/")
            self.assert_response(response)
            return response.json()
        except Exception as e:
            logger.exception(e)
            raise

    def list_applications(self, project: str) -> list[dict[str, Any]]:
        """List applications for a project.

        Args:
            project: Project slug.

        Returns:
            A list of application dicts, each containing ``slug``, ``name``.

        """
        try:
            response = self.transport.get(f"p/{project}/a/")
            self.assert_response(response)
            return response.json()
        except Exception as e:
            logger.exception(e)
            raise

    def list_members(self, project: str, distribution_list: str) -> list[dict[str, Any]]:
        """List members of a distribution list.

        Args:
            project: Project slug.
            distribution_list: Distribution list ID.

        Returns:
            A list of member dicts, each containing ``id``, ``address``,
            ``user``, ``channel``.

        """
        try:
            response = self.transport.get(f"p/{project}/d/{distribution_list}/m/")
            self.assert_response(response)
            return response.json()
        except Exception as e:
            logger.exception(e)
            raise

    def trigger(
        self,
        project: str,
        application: str,
        event: str,
        context: dict[str, str] | None = None,
        options: dict[str, str] | None = None,
        cid: str | None = None,
    ) -> dict[str, Any]:
        """Use :meth:`set_domain` + :meth:`trigger_event` instead."""
        warnings.warn(
            "trigger() is deprecated, use trigger_event() with set_domain() instead",
            DeprecationWarning,
            stacklevel=2,
        )
        try:
            if cid:
                cid = f"?cid={cid}"
            else:
                cid = ""
            url = self.transport.get_url(f"p/{project}/a/{application}/e/{event}/trigger/{cid}")
            response = self.transport.post(url, {"context": context or {}, "options": options or {}})
            if response.status_code in [404]:
                raise EventNotFoundError(f"Event not found at {url}")
            self.assert_response(response)
            return response.json()
        except Exception as e:
            logger.exception(e)
            raise

    def add_user(self, email: str, first_name: str, last_name: str, custom: "JSON | None" = None) -> "JSON":
        """Add a new user to the current organization.

        Args:
            email: User email address.
            first_name: User first name.
            last_name: User last name.
            custom: Optional dict of custom fields.

        Returns:
            The API response dict for the created user.

        """
        try:
            response = self.transport.post(
                "u/",
                {"email": email, "first_name": first_name or "", "last_name": last_name or "", "custom_fields": custom},
            )
            self.assert_response(response)
            return response.json()
        except Exception as e:
            logger.exception(e)
            raise

    def update_user(
        self,
        email: str,
        first_name: str,
        last_name: str,
        custom_fields: "JSON | None" = None,
        mode: str = JsonUpdateMode.IGNORE,
    ) -> "JSON":
        """Update an existing user in the current organization.

        Args:
            email: User email address (used as the lookup key).
            first_name: New first name.
            last_name: New last name.
            custom_fields: Optional dict of custom fields to update.
            mode: Merge mode for custom fields
                (see :class:`~bitcaster_sdk.helpers.JsonUpdateMode`).

        Returns:
            The API response dict for the updated user.

        """
        try:
            uid = urllib.parse.quote(email)
            response = self.transport.patch(
                f"u/{uid}/",
                {
                    "first_name": first_name or "",
                    "last_name": last_name or "",
                    "custom_fields": custom_fields,
                    "_mode": mode,
                },
            )
            self.assert_response(response)
            return response.json()
        except ValidationError:
            raise
        except Exception as e:
            logger.exception(e)
            raise

    def register_user(
        self,
        project: str,
        application: str,
        username: str,
        first_name: str = "",
        last_name: str = "",
        email: str = "",
        custom_fields: "JSON | None" = None,
        active: bool = True,
        addresses: list[dict[str, Any]] | None = None,
        distribution_list: str | None = None,
    ) -> "JSON":
        """Register a user as member of an application.

        Creates the user if it does not exist, stores per-application custom
        fields, optionally creates addresses, assigns them to the preferred
        channels and adds the assignments to a distribution list.

        Args:
            project: Project slug.
            application: Application slug.
            username: Username (used as the lookup key; the user is created if missing).
            first_name: User first name (only used when creating the user).
            last_name: User last name (only used when creating the user).
            email: User email address (only used when creating the user).
            custom_fields: Optional dict of per-application custom fields,
                merged into the existing membership custom fields.
            active: Whether the membership is active.
            addresses: Optional list of address dicts, each with ``value``
                (required), ``name`` and ``assign_to_preferred_channel``.
            distribution_list: Optional name of a distribution list the
                created assignments are added to.

        Returns:
            The API response dict with ``user``, ``created``, ``membership``,
            ``addresses``, ``assignments`` and ``distribution_list``.

        """
        try:
            response = self.transport.post(
                f"p/{project}/a/{application}/register/",
                {
                    "username": username,
                    "first_name": first_name or "",
                    "last_name": last_name or "",
                    "email": email or "",
                    "custom_fields": custom_fields or {},
                    "active": active,
                    "addresses": addresses or [],
                    "distribution_list": distribution_list,
                },
            )
            self.assert_response(response)
            return response.json()
        except ValidationError:
            raise
        except Exception as e:
            logger.exception(e)
            raise

    def unregister_user(self, project: str, application: str, username: str) -> "JSON":
        """Unregister a user from an application.

        Deletes the user's application membership records. Distribution list
        subscriptions are not affected.

        Args:
            project: Project slug.
            application: Application slug.
            username: Username of the user to unregister.

        Returns:
            The API response dict with ``deleted``, the number of deleted
            memberships.

        """
        try:
            uid = urllib.parse.quote(username)
            response = self.transport.post(f"p/{project}/a/{application}/unregister/{uid}/", {})
            self.assert_response(response)
            return response.json()
        except Exception as e:
            logger.exception(e)
            raise

ping()

Check connectivity with the Bitcaster server.

Returns:

Type Description
dict[str, Any]

A dict with server identifying information (e.g. token, slug).
Source code in src/bitcaster_sdk/client.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def ping(self) -> dict[str, Any]:
    """Check connectivity with the Bitcaster server.

    Returns:
        A dict with server identifying information (e.g. ``token``, ``slug``).

    """
    try:
        response = self.transport.get("/api/system/ping/")
        self.assert_response(response)
        return response.json()
    except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
        raise ConnectionError(f"Connection Error: {self.api_url}") from e
    except Exception as e:
        logger.exception(e)
        raise

list_projects()

List projects in the current organization.

Returns:

Type Description
list[dict[str, Any]]

A list of project dicts, each containing slug, name,
list[dict[str, Any]]

applications, lists, channels.
Source code in src/bitcaster_sdk/client.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def list_projects(self) -> list[dict[str, Any]]:
    """List projects in the current organization.

    Returns:
        A list of project dicts, each containing ``slug``, ``name``,
        ``applications``, ``lists``, ``channels``.

    """
    try:
        response = self.transport.get("p/")
        self.assert_response(response)
        return response.json()
    except Exception as e:
        logger.exception(e)
        raise

list_applications(project)

List applications for a project.

Parameters:

Name Type Description Default
project str

Project slug.

 required

Returns:

Type Description
list[dict[str, Any]]

A list of application dicts, each containing slug, name.
Source code in src/bitcaster_sdk/client.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def list_applications(self, project: str) -> list[dict[str, Any]]:
    """List applications for a project.

    Args:
        project: Project slug.

    Returns:
        A list of application dicts, each containing ``slug``, ``name``.

    """
    try:
        response = self.transport.get(f"p/{project}/a/")
        self.assert_response(response)
        return response.json()
    except Exception as e:
        logger.exception(e)
        raise

list_events(project, application)

List events for a given project and application.

Parameters:

Name Type Description Default
project str

Project slug.

 required
application str

Application slug.

 required

Returns:

Type Description
list[dict[str, Any]]

A list of event dicts, each containing name, slug,
list[dict[str, Any]]

active, locked, description.
Source code in src/bitcaster_sdk/client.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def list_events(self, project: str, application: str) -> list[dict[str, Any]]:
    """List events for a given project and application.

    Args:
        project: Project slug.
        application: Application slug.

    Returns:
        A list of event dicts, each containing ``name``, ``slug``,
        ``active``, ``locked``, ``description``.

    """
    try:
        response = self.transport.get(f"p/{project}/a/{application}/e/")
        self.assert_response(response)
        return response.json()
    except Exception as e:
        logger.exception(e)
        raise e

list_distribution_lists(project)

List distribution lists for a project.

Parameters:

Name Type Description Default
project str

Project slug.

 required

Returns:

Type Description
list[dict[str, Any]]

A list of distribution list dicts.
Source code in src/bitcaster_sdk/client.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def list_distribution_lists(self, project: str) -> list[dict[str, Any]]:
    """List distribution lists for a project.

    Args:
        project: Project slug.

    Returns:
        A list of distribution list dicts.

    """
    try:
        response = self.transport.get(f"p/{project}/d/")
        self.assert_response(response)
        return response.json()
    except Exception as e:
        logger.exception(e)
        raise

list_members(project, distribution_list)

List members of a distribution list.

Parameters:

Name Type Description Default
project str

Project slug.

 required
distribution_list str

Distribution list ID.

 required

Returns:

Type Description
list[dict[str, Any]]

A list of member dicts, each containing id, address,
list[dict[str, Any]]

user, channel.
Source code in src/bitcaster_sdk/client.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def list_members(self, project: str, distribution_list: str) -> list[dict[str, Any]]:
    """List members of a distribution list.

    Args:
        project: Project slug.
        distribution_list: Distribution list ID.

    Returns:
        A list of member dicts, each containing ``id``, ``address``,
        ``user``, ``channel``.

    """
    try:
        response = self.transport.get(f"p/{project}/d/{distribution_list}/m/")
        self.assert_response(response)
        return response.json()
    except Exception as e:
        logger.exception(e)
        raise

list_users()

List users in the current organization.

Returns:

Type Description
list[dict[str, Any]]

A list of user dicts, each containing email, username,
list[dict[str, Any]]

is_active, locked, etc.
Source code in src/bitcaster_sdk/client.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def list_users(self) -> list[dict[str, Any]]:
    """List users in the current organization.

    Returns:
        A list of user dicts, each containing ``email``, ``username``,
        ``is_active``, ``locked``, etc.

    """
    try:
        response = self.transport.get("u/")
        self.assert_response(response)
        return response.json()
    except Exception as e:
        logger.exception(e)
        raise

add_user(email, first_name, last_name, custom=None)

Add a new user to the current organization.

Parameters:

Name Type Description Default
email str

User email address.

 required
first_name str

User first name.

 required
last_name str

User last name.

 required
custom 'JSON | None'

Optional dict of custom fields.

 None

Returns:

Type Description
'JSON'

The API response dict for the created user.
Source code in src/bitcaster_sdk/client.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def add_user(self, email: str, first_name: str, last_name: str, custom: "JSON | None" = None) -> "JSON":
    """Add a new user to the current organization.

    Args:
        email: User email address.
        first_name: User first name.
        last_name: User last name.
        custom: Optional dict of custom fields.

    Returns:
        The API response dict for the created user.

    """
    try:
        response = self.transport.post(
            "u/",
            {"email": email, "first_name": first_name or "", "last_name": last_name or "", "custom_fields": custom},
        )
        self.assert_response(response)
        return response.json()
    except Exception as e:
        logger.exception(e)
        raise

update_user(email, first_name, last_name, custom_fields=None, mode=JsonUpdateMode.IGNORE)

Update an existing user in the current organization.

Parameters:

Name Type Description Default
email str

User email address (used as the lookup key).

 required
first_name str

New first name.

 required
last_name str

New last name.

 required
custom_fields 'JSON | None'

Optional dict of custom fields to update.

 None
mode str

Merge mode for custom fields (see :class:~bitcaster_sdk.helpers.JsonUpdateMode).

 IGNORE

Returns:

Type Description
'JSON'

The API response dict for the updated user.
Source code in src/bitcaster_sdk/client.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def update_user(
    self,
    email: str,
    first_name: str,
    last_name: str,
    custom_fields: "JSON | None" = None,
    mode: str = JsonUpdateMode.IGNORE,
) -> "JSON":
    """Update an existing user in the current organization.

    Args:
        email: User email address (used as the lookup key).
        first_name: New first name.
        last_name: New last name.
        custom_fields: Optional dict of custom fields to update.
        mode: Merge mode for custom fields
            (see :class:`~bitcaster_sdk.helpers.JsonUpdateMode`).

    Returns:
        The API response dict for the updated user.

    """
    try:
        uid = urllib.parse.quote(email)
        response = self.transport.patch(
            f"u/{uid}/",
            {
                "first_name": first_name or "",
                "last_name": last_name or "",
                "custom_fields": custom_fields,
                "_mode": mode,
            },
        )
        self.assert_response(response)
        return response.json()
    except ValidationError:
        raise
    except Exception as e:
        logger.exception(e)
        raise

register_user(project, application, username, first_name='', last_name='', email='', custom_fields=None, active=True, addresses=None, distribution_list=None)

Register a user as member of an application.

Creates the user if it does not exist, stores per-application custom fields, optionally creates addresses, assigns them to the preferred channels and adds the assignments to a distribution list.

Parameters:

Name Type Description Default
project str

Project slug.

 required
application str

Application slug.

 required
username str

Username (used as the lookup key; the user is created if missing).

 required
first_name str

User first name (only used when creating the user).

 ''
last_name str

User last name (only used when creating the user).

 ''
email str

User email address (only used when creating the user).

 ''
custom_fields 'JSON | None'

Optional dict of per-application custom fields, merged into the existing membership custom fields.

 None
active bool

Whether the membership is active.

 True
addresses list[dict[str, Any]] | None

Optional list of address dicts, each with value (required), name and assign_to_preferred_channel.

 None
distribution_list str | None

Optional name of a distribution list the created assignments are added to.

 None

Returns:

Type Description
'JSON'

The API response dict with user, created, membership,
'JSON'

addresses, assignments and distribution_list.
Source code in src/bitcaster_sdk/client.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
def register_user(
    self,
    project: str,
    application: str,
    username: str,
    first_name: str = "",
    last_name: str = "",
    email: str = "",
    custom_fields: "JSON | None" = None,
    active: bool = True,
    addresses: list[dict[str, Any]] | None = None,
    distribution_list: str | None = None,
) -> "JSON":
    """Register a user as member of an application.

    Creates the user if it does not exist, stores per-application custom
    fields, optionally creates addresses, assigns them to the preferred
    channels and adds the assignments to a distribution list.

    Args:
        project: Project slug.
        application: Application slug.
        username: Username (used as the lookup key; the user is created if missing).
        first_name: User first name (only used when creating the user).
        last_name: User last name (only used when creating the user).
        email: User email address (only used when creating the user).
        custom_fields: Optional dict of per-application custom fields,
            merged into the existing membership custom fields.
        active: Whether the membership is active.
        addresses: Optional list of address dicts, each with ``value``
            (required), ``name`` and ``assign_to_preferred_channel``.
        distribution_list: Optional name of a distribution list the
            created assignments are added to.

    Returns:
        The API response dict with ``user``, ``created``, ``membership``,
        ``addresses``, ``assignments`` and ``distribution_list``.

    """
    try:
        response = self.transport.post(
            f"p/{project}/a/{application}/register/",
            {
                "username": username,
                "first_name": first_name or "",
                "last_name": last_name or "",
                "email": email or "",
                "custom_fields": custom_fields or {},
                "active": active,
                "addresses": addresses or [],
                "distribution_list": distribution_list,
            },
        )
        self.assert_response(response)
        return response.json()
    except ValidationError:
        raise
    except Exception as e:
        logger.exception(e)
        raise

unregister_user(project, application, username)

Unregister a user from an application.

Deletes the user's application membership records. Distribution list subscriptions are not affected.

Parameters:

Name Type Description Default
project str

Project slug.

 required
application str

Application slug.

 required
username str

Username of the user to unregister.

 required

Returns:

Type Description
'JSON'

The API response dict with deleted, the number of deleted
'JSON'

memberships.
Source code in src/bitcaster_sdk/client.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def unregister_user(self, project: str, application: str, username: str) -> "JSON":
    """Unregister a user from an application.

    Deletes the user's application membership records. Distribution list
    subscriptions are not affected.

    Args:
        project: Project slug.
        application: Application slug.
        username: Username of the user to unregister.

    Returns:
        The API response dict with ``deleted``, the number of deleted
        memberships.

    """
    try:
        uid = urllib.parse.quote(username)
        response = self.transport.post(f"p/{project}/a/{application}/unregister/{uid}/", {})
        self.assert_response(response)
        return response.json()
    except Exception as e:
        logger.exception(e)
        raise