broadcaster Library

0.3.1 · active · verified Fri Apr 17

Broadcaster provides a simple, asynchronous publish/subscribe interface for Python 3.8+, allowing for broadcast channels across various backends like Redis, PostgreSQL, and Kafka. The current version is 0.3.1. It's actively maintained by the Encode organization, known for projects like Starlette and Uvicorn, suggesting a steady, incremental release cadence.

Common errors

```
ModuleNotFoundError: No module named 'redis'
```
cause The specific backend dependency was not installed. The base `broadcaster` package only includes the in-memory backend.

fix Install the correct optional dependency for your chosen backend, e.g., `pip install "broadcaster[redis]"` for Redis, `pip install "broadcaster[postgres]"` for PostgreSQL, or `pip install "broadcaster[kafka]"` for Kafka.
```
RuntimeError: Event loop is closed
```
cause This typically occurs in asyncio when attempting to run async code after the event loop has been shut down, or when mismanaging multiple event loops.

fix Ensure that `asyncio.run()` is used correctly as the entry point for your async application, or if integrating with a larger framework, verify its event loop management. Avoid explicitly closing loops if `asyncio.run()` is managing it.
```
broadcaster.exceptions.ConnectionError: Unable to connect to backend at ...
```
cause The specified backend service (e.g., Redis server, PostgreSQL database, Kafka broker) is not running, is inaccessible from the application's host, or the `BROADCAST_URL` is incorrect.

fix Verify that the backend service is running, is reachable on the specified host and port, and that the `BROADCAST_URL` string is correctly formatted (e.g., `redis://localhost:6379`). Check firewall rules or network configurations if connecting to a remote service.

Warnings

gotcha Backend dependencies (e.g., Redis, PostgreSQL, Kafka) are optional and must be explicitly installed for each chosen backend. Installing just `broadcaster` will only provide the in-memory backend.
Fix: Install the specific backend dependency using extras, e.g., `pip install "broadcaster[redis]"` for Redis, or `pip install "broadcaster[postgres]"` for PostgreSQL.
gotcha Always ensure to `await broadcaster.connect()` before publishing or subscribing, and `await broadcaster.disconnect()` afterwards. Forgetting these can lead to resource leaks or operations failing due to a lack of an active connection.
Fix: Add `await broadcaster.connect()` at the start of your Broadcaster operations and `await broadcaster.disconnect()` when done, or use context managers for subscriptions where applicable.
gotcha Broadcaster is an asyncio-first library. All interaction methods like `publish()` and `subscribe()` are coroutines and must be `await`ed. Mixing synchronous and asynchronous code without proper handling will lead to `TypeError: object 'coroutine' can't be used in 'await' expression` or other runtime errors.
Fix: Ensure all Broadcaster operations are awaited within an `async` function and the application's event loop is properly managed (e.g., using `asyncio.run()` for top-level execution).

Install

pip install broadcaster Base installation
pip install "broadcaster[redis]" "broadcaster[postgres]" "broadcaster[kafka]" With all optional backends

Imports

Broadcaster
```
from broadcaster import Broadcaster
```

Quickstart

This quickstart demonstrates setting up an in-memory (default) or environment-variable-configured Broadcaster. It connects, publishes a message to a channel, and then subscribes and retrieves that message. Remember to set the `BROADCAST_URL` environment variable (e.g., `redis://localhost:6379`) to use a different backend, and ensure the corresponding optional dependency is installed.

import asyncio
import os
from broadcaster import Broadcaster

BROADCAST_URL = os.environ.get("BROADCAST_URL", "memory://")

async def main():
    # Initialize broadcaster with a URL, defaults to in-memory if env var is not set
    broadcaster = Broadcaster(BROADCAST_URL)
    
    # Establish connection to the backend
    await broadcaster.connect()
    
    async with broadcaster.subscribe("chat") as subscriber:
        # Publish a message
        await broadcaster.publish("chat", "Hello from broadcaster!")
        print(f"Published: Hello from broadcaster!")
        
        # Receive a message
        message = await subscriber.get()
        print(f"Received: {message}")

    # Disconnect from the backend
    await broadcaster.disconnect()

if __name__ == "__main__":
    asyncio.run(main())

view raw JSON →