Async Client

The AsyncClient provides a non-blocking interface to the Bitcaster API. Every API method returns a [concurrent.futures.Future][], allowing your application to continue executing while the request is processed in a background thread.

For shared constructor arguments, properties, and inherited methods see the Sync Client reference.

Quick Start

from bitcaster_sdk.async_client import AsyncClient

client = AsyncClient("https://key@app.bitcaster.io/api/o/ORG/")
future = client.ping()
result = future.result(timeout=10)

Motivation

The sync Client blocks the calling thread on every HTTP request. AsyncClient delegates requests to a daemon-thread [BackgroundWorker][bitcaster_sdk.async_worker.BackgroundWorker], freeing the caller to do other work while the HTTP round-trip completes.

This is especially useful in:

• Web applications (e.g. Django views) where you want to return a response without waiting for the Bitcaster API
• Event-driven systems where you fire notifications and move on
• High-throughput pipelines that send many independent requests concurrently

Architecture

┌──────────────┐     submit()     ┌──────────────────┐
│ AsyncClient  │ ───────────────► │ AsyncTransport   │
│              │                  │                  │
│  ping()      │   Future[Resp]   │  BackgroundWorker │
│  trigger()   │ ◄─────────────── │  (daemon thread)  │
│  list_*()    │                  │  ┌────────────┐   │
│  add_user()  │                  │  │  Queue     │   │
│  update_user()│                  │  │  (RLock)   │   │
└──────────────┘                  │  └────────────┘   │
                                  └──────────────────┘
                                          │
                                          ▼
                                  ┌──────────────────┐
                                  │  HTTP Session    │
                                  │  (requests)      │
                                  └──────────────────┘

Usage

Initialisation

client = AsyncClient("https://key@app.bitcaster.io/api/o/ORG/")

The base_url format is identical to Client:

https://<API_KEY>@<SERVER>/api/o/<organization_slug>/

Non-blocking calls

Every method returns a Future. Call .result(timeout) to get the actual response:

# Fire and collect later
ping_future = client.ping()
events_future = client.list_events("my-project", "my-app")

# Block only when you need the result
print(ping_future.result(timeout=10))
print(events_future.result(timeout=10))

Context manager

with AsyncClient("https://key@app.bitcaster.io/api/o/ORG/") as client:
    result = client.ping().result(timeout=10)

The context manager calls close() on exit, which flushes pending work and shuts down the background thread.

Flushing and shutdown

# Wait for all queued requests to complete (default 10s)
client.flush()

# Shut down the background thread
client.close()

API Reference

Bases: AbstractClient

Non-blocking Bitcaster client.

Every API method returns a :class:~concurrent.futures.Future and executes the HTTP request on a background thread, leaving the caller free to do other work.

See :class:~bitcaster_sdk.client.Client for constructor arguments, properties, and shared method documentation. Only overridden methods and AsyncClient-specific methods are documented below.

Source code in src/bitcaster_sdk/async_client.py

class AsyncClient(AbstractClient):
    """Non-blocking Bitcaster client.

    Every API method returns a :class:`~concurrent.futures.Future` and
    executes the HTTP request on a background thread, leaving the caller
    free to do other work.

    See :class:`~bitcaster_sdk.client.Client` for constructor arguments,
    properties, and shared method documentation. Only overridden methods
    and AsyncClient-specific methods are documented below.
    """

    _transport_class: type[AbstractTransport] = AsyncTransport

    def _submit(self, fn: Any, *args: Any, **kwargs: Any) -> Future[Any]:
        if self.transport is None:
            future: Future[Any] = Future()
            future.set_exception(RuntimeError("client not initialized"))
            return future
        return self.transport.submit(fn, *args, **kwargs)

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

        Returns:
            A Future that resolves to a dict with server information.

        """
        url = self.transport.get_url("/api/system/ping/") if self.transport else ""

        def _call() -> dict[str, Any]:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            try:
                self.transport.last_url = url
                response = self.transport.session.get(url, timeout=self.transport.timeout)
                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

        return self._submit(_call)

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

        Returns:
            A Future that resolves to a list of event dicts.

        """
        url = self.transport.get_url(f"p/{project}/a/{application}/e/") if self.transport else ""

        def _call() -> list[dict[str, Any]]:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            self.transport.last_url = url
            response = self.transport.session.get(url, timeout=self.transport.timeout)
            self.assert_response(response)
            return response.json()

        return self._submit(_call)

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

        Returns:
            A Future that resolves to a list of user dicts.

        """
        url = self.transport.get_url("u/") if self.transport else ""

        def _call() -> list[dict[str, Any]]:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            self.transport.last_url = url
            response = self.transport.session.get(url, timeout=self.transport.timeout)
            self.assert_response(response)
            return response.json()

        return self._submit(_call)

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

        Returns:
            A Future that resolves to a list of distribution list dicts.

        """
        url = self.transport.get_url(f"p/{project}/d/") if self.transport else ""

        def _call() -> list[dict[str, Any]]:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            self.transport.last_url = url
            response = self.transport.session.get(url, timeout=self.transport.timeout)
            self.assert_response(response)
            return response.json()

        return self._submit(_call)

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

        Returns:
            A Future that resolves to a list of project dicts.

        """
        url = self.transport.get_url("p/") if self.transport else ""

        def _call() -> list[dict[str, Any]]:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            self.transport.last_url = url
            response = self.transport.session.get(url, timeout=self.transport.timeout)
            self.assert_response(response)
            return response.json()

        return self._submit(_call)

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

        Returns:
            A Future that resolves to a list of application dicts.

        """
        url = self.transport.get_url(f"p/{project}/a/") if self.transport else ""

        def _call() -> list[dict[str, Any]]:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            self.transport.last_url = url
            response = self.transport.session.get(url, timeout=self.transport.timeout)
            self.assert_response(response)
            return response.json()

        return self._submit(_call)

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

        Returns:
            A Future that resolves to a list of member dicts.

        """
        url = self.transport.get_url(f"p/{project}/d/{distribution_list}/m/") if self.transport else ""

        def _call() -> list[dict[str, Any]]:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            self.transport.last_url = url
            response = self.transport.session.get(url, timeout=self.transport.timeout)
            self.assert_response(response)
            return response.json()

        return self._submit(_call)

    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,
    ) -> Future[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,
        )

        def _call() -> dict[str, Any]:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            url = self.transport.get_url(
                f"p/{project}/a/{application}/e/{event}/trigger/{'?cid=' + cid if cid else ''}"
            )
            self.transport.last_url = url
            with self.transport.with_headers({"Content-Type": "application/json"}):
                response = self.transport.session.post(
                    url,
                    json={"context": context or {}, "options": options or {}},
                    timeout=self.transport.timeout,
                )
            if response.status_code in [404]:
                raise EventNotFoundError(f"Event not found at {url}")
            self.assert_response(response)
            return response.json()

        return self._submit(_call)

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

        Returns:
            A Future that resolves to the created user dict.

        """

        def _call() -> JSON:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            url = self.transport.get_url("u/")
            self.transport.last_url = url
            with self.transport.with_headers({"Content-Type": "application/json"}):
                response = self.transport.session.post(
                    url,
                    json={
                        "email": email,
                        "first_name": first_name or "",
                        "last_name": last_name or "",
                        "custom_fields": custom,
                    },
                    timeout=self.transport.timeout,
                )
            self.assert_response(response)
            return response.json()

        return self._submit(_call)

    def update_user(
        self,
        email: str,
        first_name: str,
        last_name: str,
        custom_fields: JSON | None = None,
        mode: str = JsonUpdateMode.IGNORE,
    ) -> Future[JSON]:
        """Update an existing user (async).

        Returns:
            A Future that resolves to the updated user dict.

        """
        uid = urllib.parse.quote(email)

        def _call() -> JSON:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            url = self.transport.get_url(f"u/{uid}/")
            self.transport.last_url = url
            with self.transport.with_headers({"Content-Type": "application/json"}):
                response = self.transport.session.patch(
                    url,
                    json={
                        "first_name": first_name or "",
                        "last_name": last_name or "",
                        "custom_fields": custom_fields,
                        "_mode": mode,
                    },
                    timeout=self.transport.timeout,
                )
            self.assert_response(response)
            return response.json()

        return self._submit(_call)

    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,
    ) -> Future[JSON]:
        """Register a user as member of an application (async).

        Returns:
            A Future that resolves to the registration response dict.

        """

        def _call() -> JSON:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            url = self.transport.get_url(f"p/{project}/a/{application}/register/")
            self.transport.last_url = url
            with self.transport.with_headers({"Content-Type": "application/json"}):
                response = self.transport.session.post(
                    url,
                    json={
                        "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,
                    },
                    timeout=self.transport.timeout,
                )
            self.assert_response(response)
            return response.json()

        return self._submit(_call)

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

        Returns:
            A Future that resolves to a dict with ``deleted``, the number of
            deleted memberships.

        """
        uid = urllib.parse.quote(username)

        def _call() -> JSON:
            if self.transport is None:
                raise RuntimeError("client not initialized")
            url = self.transport.get_url(f"p/{project}/a/{application}/unregister/{uid}/")
            self.transport.last_url = url
            with self.transport.with_headers({"Content-Type": "application/json"}):
                response = self.transport.session.post(url, json={}, timeout=self.transport.timeout)
            self.assert_response(response)
            return response.json()

        return self._submit(_call)

    def flush(self, timeout: float | None = None) -> bool:
        """Wait for all queued requests to complete.

        Args:
            timeout: Maximum seconds to wait. Falls back to
                ``shutdown_timeout`` (default 10).

        Returns:
            ``True`` if all requests completed, ``False`` on timeout.

        """
        if self.transport is None:
            return True
        if timeout is None:
            timeout = self.options.get("shutdown_timeout", 10)
        return self.transport.flush(timeout)

    def close(self, timeout: float | None = None) -> None:
        """Flush pending work and shut down the background worker.

        The client is unusable after calling this method.

        Args:
            timeout: Maximum seconds to wait for pending requests.

        """
        if self.transport is None:
            return
        self.flush(timeout)
        self.transport.kill()
        self.transport = None

    def __enter__(self) -> AsyncClient:
        """Enter the runtime context."""
        return self

    def __exit__(self, *args: Any) -> None:
        """Exit the runtime context and shut down the background worker."""
        self.close()

ping()

Check connectivity with the Bitcaster server (async).

Returns:

Type	Description
Future[dict[str, Any]]	A Future that resolves to a dict with server information.

Source code in src/bitcaster_sdk/async_client.py

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

    Returns:
        A Future that resolves to a dict with server information.

    """
    url = self.transport.get_url("/api/system/ping/") if self.transport else ""

    def _call() -> dict[str, Any]:
        if self.transport is None:
            raise RuntimeError("client not initialized")
        try:
            self.transport.last_url = url
            response = self.transport.session.get(url, timeout=self.transport.timeout)
            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

    return self._submit(_call)

list_events(project, application)

List events for a given project and application (async).

Returns:

Type	Description
Future[list[dict[str, Any]]]	A Future that resolves to a list of event dicts.

Source code in src/bitcaster_sdk/async_client.py

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

    Returns:
        A Future that resolves to a list of event dicts.

    """
    url = self.transport.get_url(f"p/{project}/a/{application}/e/") if self.transport else ""

    def _call() -> list[dict[str, Any]]:
        if self.transport is None:
            raise RuntimeError("client not initialized")
        self.transport.last_url = url
        response = self.transport.session.get(url, timeout=self.transport.timeout)
        self.assert_response(response)
        return response.json()

    return self._submit(_call)

list_users()

List organization users (async).

Returns:

Type	Description
Future[list[dict[str, Any]]]	A Future that resolves to a list of user dicts.

Source code in src/bitcaster_sdk/async_client.py

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

    Returns:
        A Future that resolves to a list of user dicts.

    """
    url = self.transport.get_url("u/") if self.transport else ""

    def _call() -> list[dict[str, Any]]:
        if self.transport is None:
            raise RuntimeError("client not initialized")
        self.transport.last_url = url
        response = self.transport.session.get(url, timeout=self.transport.timeout)
        self.assert_response(response)
        return response.json()

    return self._submit(_call)

list_projects()

List organization projects (async).

Returns:

Type	Description
Future[list[dict[str, Any]]]	A Future that resolves to a list of project dicts.

Source code in src/bitcaster_sdk/async_client.py

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

    Returns:
        A Future that resolves to a list of project dicts.

    """
    url = self.transport.get_url("p/") if self.transport else ""

    def _call() -> list[dict[str, Any]]:
        if self.transport is None:
            raise RuntimeError("client not initialized")
        self.transport.last_url = url
        response = self.transport.session.get(url, timeout=self.transport.timeout)
        self.assert_response(response)
        return response.json()

    return self._submit(_call)

list_applications(project)

List applications for a project (async).

Returns:

Type	Description
Future[list[dict[str, Any]]]	A Future that resolves to a list of application dicts.

Source code in src/bitcaster_sdk/async_client.py

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

    Returns:
        A Future that resolves to a list of application dicts.

    """
    url = self.transport.get_url(f"p/{project}/a/") if self.transport else ""

    def _call() -> list[dict[str, Any]]:
        if self.transport is None:
            raise RuntimeError("client not initialized")
        self.transport.last_url = url
        response = self.transport.session.get(url, timeout=self.transport.timeout)
        self.assert_response(response)
        return response.json()

    return self._submit(_call)

list_distribution_lists(project)

List distribution lists for a project (async).

Returns:

Type	Description
Future[list[dict[str, Any]]]	A Future that resolves to a list of distribution list dicts.

Source code in src/bitcaster_sdk/async_client.py

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

    Returns:
        A Future that resolves to a list of distribution list dicts.

    """
    url = self.transport.get_url(f"p/{project}/d/") if self.transport else ""

    def _call() -> list[dict[str, Any]]:
        if self.transport is None:
            raise RuntimeError("client not initialized")
        self.transport.last_url = url
        response = self.transport.session.get(url, timeout=self.transport.timeout)
        self.assert_response(response)
        return response.json()

    return self._submit(_call)

list_members(project, distribution_list)

List distribution list members (async).

Returns:

Type	Description
Future[list[dict[str, Any]]]	A Future that resolves to a list of member dicts.

Source code in src/bitcaster_sdk/async_client.py

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

    Returns:
        A Future that resolves to a list of member dicts.

    """
    url = self.transport.get_url(f"p/{project}/d/{distribution_list}/m/") if self.transport else ""

    def _call() -> list[dict[str, Any]]:
        if self.transport is None:
            raise RuntimeError("client not initialized")
        self.transport.last_url = url
        response = self.transport.session.get(url, timeout=self.transport.timeout)
        self.assert_response(response)
        return response.json()

    return self._submit(_call)

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

Add a user to the current organization (async).

Returns:

Type	Description
Future[JSON]	A Future that resolves to the created user dict.

Source code in src/bitcaster_sdk/async_client.py

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

    Returns:
        A Future that resolves to the created user dict.

    """

    def _call() -> JSON:
        if self.transport is None:
            raise RuntimeError("client not initialized")
        url = self.transport.get_url("u/")
        self.transport.last_url = url
        with self.transport.with_headers({"Content-Type": "application/json"}):
            response = self.transport.session.post(
                url,
                json={
                    "email": email,
                    "first_name": first_name or "",
                    "last_name": last_name or "",
                    "custom_fields": custom,
                },
                timeout=self.transport.timeout,
            )
        self.assert_response(response)
        return response.json()

    return self._submit(_call)

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

Update an existing user (async).

Returns:

Type	Description
Future[JSON]	A Future that resolves to the updated user dict.

Source code in src/bitcaster_sdk/async_client.py

def update_user(
    self,
    email: str,
    first_name: str,
    last_name: str,
    custom_fields: JSON | None = None,
    mode: str = JsonUpdateMode.IGNORE,
) -> Future[JSON]:
    """Update an existing user (async).

    Returns:
        A Future that resolves to the updated user dict.

    """
    uid = urllib.parse.quote(email)

    def _call() -> JSON:
        if self.transport is None:
            raise RuntimeError("client not initialized")
        url = self.transport.get_url(f"u/{uid}/")
        self.transport.last_url = url
        with self.transport.with_headers({"Content-Type": "application/json"}):
            response = self.transport.session.patch(
                url,
                json={
                    "first_name": first_name or "",
                    "last_name": last_name or "",
                    "custom_fields": custom_fields,
                    "_mode": mode,
                },
                timeout=self.transport.timeout,
            )
        self.assert_response(response)
        return response.json()

    return self._submit(_call)

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

Returns:

Type	Description
Future[JSON]	A Future that resolves to the registration response dict.

Source code in src/bitcaster_sdk/async_client.py

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,
) -> Future[JSON]:
    """Register a user as member of an application (async).

    Returns:
        A Future that resolves to the registration response dict.

    """

    def _call() -> JSON:
        if self.transport is None:
            raise RuntimeError("client not initialized")
        url = self.transport.get_url(f"p/{project}/a/{application}/register/")
        self.transport.last_url = url
        with self.transport.with_headers({"Content-Type": "application/json"}):
            response = self.transport.session.post(
                url,
                json={
                    "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,
                },
                timeout=self.transport.timeout,
            )
        self.assert_response(response)
        return response.json()

    return self._submit(_call)

unregister_user(project, application, username)

Unregister a user from an application (async).

Returns:

Type	Description
Future[JSON]	A Future that resolves to a dict with deleted, the number of
Future[JSON]	deleted memberships.

Source code in src/bitcaster_sdk/async_client.py

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

    Returns:
        A Future that resolves to a dict with ``deleted``, the number of
        deleted memberships.

    """
    uid = urllib.parse.quote(username)

    def _call() -> JSON:
        if self.transport is None:
            raise RuntimeError("client not initialized")
        url = self.transport.get_url(f"p/{project}/a/{application}/unregister/{uid}/")
        self.transport.last_url = url
        with self.transport.with_headers({"Content-Type": "application/json"}):
            response = self.transport.session.post(url, json={}, timeout=self.transport.timeout)
        self.assert_response(response)
        return response.json()

    return self._submit(_call)

flush(timeout=None)

Wait for all queued requests to complete.

Parameters:

Name	Type	Description	Default
timeout	float | None	Maximum seconds to wait. Falls back to shutdown_timeout (default 10).	None

Returns:

Type	Description
bool	True if all requests completed, False on timeout.

Source code in src/bitcaster_sdk/async_client.py

def flush(self, timeout: float | None = None) -> bool:
    """Wait for all queued requests to complete.

    Args:
        timeout: Maximum seconds to wait. Falls back to
            ``shutdown_timeout`` (default 10).

    Returns:
        ``True`` if all requests completed, ``False`` on timeout.

    """
    if self.transport is None:
        return True
    if timeout is None:
        timeout = self.options.get("shutdown_timeout", 10)
    return self.transport.flush(timeout)

close(timeout=None)

Flush pending work and shut down the background worker.

The client is unusable after calling this method.

Parameters:

Name	Type	Description	Default
timeout	float | None	Maximum seconds to wait for pending requests.	None

Source code in src/bitcaster_sdk/async_client.py

def close(self, timeout: float | None = None) -> None:
    """Flush pending work and shut down the background worker.

    The client is unusable after calling this method.

    Args:
        timeout: Maximum seconds to wait for pending requests.

    """
    if self.transport is None:
        return
    self.flush(timeout)
    self.transport.kill()
    self.transport = None

Examples

Trigger an event asynchronously

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

# Do other work while the request is in-flight
process_local_task()

# Get the result (blocks if not yet available)
response = future.result(timeout=10)

Concurrent list operations

client = AsyncClient("https://key@app.bitcaster.io/api/o/ORG/")

# Fire all three requests concurrently
users_fut = client.list_users()
projects_fut = client.list_projects()
events_fut = client.list_events("my-project", "my-app")

# Collect results
users = users_fut.result(timeout=10)
projects = projects_fut.result(timeout=10)
events = events_fut.result(timeout=10)

Thread safety

AsyncClient is built on a vendored [Queue][bitcaster_sdk.async_queue.Queue] with [RLock][] and is safe to use from multiple threads. The background worker is fork-safe — if the process forks, a new worker thread is started automatically.