Small publish/subscribe event bus with typed events, inspired by the CherryPy bus.

Events are plain classes that carry the payload type and the listener return type, so event payloads and results can be statically checked with mypy (the package ships a py.typed marker).

import trolleybus

class UserLoggedIn(trolleybus.Event[str, None]):
    pass

bus = trolleybus.EventBus()

@bus.subscribe(UserLoggedIn)
def greet(username: str) -> None:
    print(f'Hello, {username}!')

bus.broadcast(UserLoggedIn, 'kate')

Features

  • Typed event payloads and listener results (Event[Payload, Result])

  • Synchronous dispatch ordered by priority (higher value runs first, subscription order breaks ties)

  • broadcast, broadcast_nothrow, send_one, send_any emission styles

  • Class-based Subscriber with declarative @subscribe handlers bound to the bus lifecycle (OnStart / OnExit)

  • Emitter / EmitterMixin facades for components that only emit

  • Thread-safe subscription management and broadcasting

Installation

$ pip install trolleybus

Usage

Events

An event defines the payload type passed to the listeners and the type of the value they return. Events are never instantiated.

import trolleybus

class UserLoggedIn(trolleybus.Event[str, None]):
    """payload: username, listeners return nothing"""

class ResolveConfig(trolleybus.Event[str, dict]):
    """payload: config key, listeners return the resolved value"""

If no name is given, the class name is used.

Subscribing

bus = trolleybus.EventBus()

# decorator form
@bus.subscribe(UserLoggedIn, priority=60)
def greet(username: str) -> None:
    ...

# direct form
def audit(username: str) -> None:
    ...

bus.subscribe(UserLoggedIn, audit, priority=40)
bus.unsubscribe(UserLoggedIn, audit)

Subscribing the same listener twice only updates its priority.

Emitting

# call all listeners, collect results, propagate listener errors
bus.broadcast(UserLoggedIn, 'kate')

# call all listeners, capture errors as ListenerResult(value, error)
for result in bus.broadcast_nothrow(UserLoggedIn, 'kate'):
    if not result.ok:
        print('listener failed:', result.error)

# call only the highest priority listener (NoListenersError if none)
config = bus.send_one(ResolveConfig, 'database.url')

# call listeners until one returns a non-None value
config = bus.send_any(ResolveConfig, 'database.url')

The bus exposes two lifecycle events: bus.start() emits OnStart and then OnStarted, bus.stop() emits OnExit.

Class-based subscribers

class Metrics(trolleybus.Subscriber):
    @trolleybus.subscribe(UserLoggedIn)
    def on_user_logged_in(self, username: str) -> None:
        ...

metrics = Metrics(bus)
bus.start()  # handlers attach
bus.stop()   # handlers detach; they re-attach on the next start

Emitters

class LoginService(trolleybus.Subscriber, trolleybus.EmitterMixin):
    def login(self, username: str) -> None:
        self.broadcast(UserLoggedIn, username)

A standalone emitter also exists: emitter = trolleybus.Emitter(bus).

Development

$ poetry install
$ poetry run pytest tests
$ poetry run mypy trolleybus
$ poetry run pylint trolleybus

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Latest release

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for blanebf from gravatar.com blanebf

Credits

Author: Pavel 'Blane' Tuchin

Requires

Python <4.0, >=3.10

Classifiers

Report project as malware

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

trolleybus-0.2.0.tar.gz (7.3 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

trolleybus-0.2.0-py3-none-any.whl (9.4 kB view details)

Uploaded Python 3

File details

Details for the file trolleybus-0.2.0.tar.gz.

File metadata

  • Download URL: trolleybus-0.2.0.tar.gz
  • Upload date:
  • Size: 7.3 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/7.0.0 CPython/3.13.14

File hashes

Hashes for trolleybus-0.2.0.tar.gz
Algorithm Hash digest
SHA256 74a7ee47b74dec461d27b932f0af26a34fd4c64e3ebdc3967346ad9e7a83032e
MD5 86897414165bbcd2b0a4833b353f9473
BLAKE2b-256 9b9e60446a3ea3ed01dd2633ab4f281b7fd73ba77b756881ea695bf81ca68d14

See more details on using hashes here.

Provenance

The following attestation bundles were made for trolleybus-0.2.0.tar.gz:

Publisher: publish.yml on blanebf/trolleybus

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file trolleybus-0.2.0-py3-none-any.whl.

File metadata

  • Download URL: trolleybus-0.2.0-py3-none-any.whl
  • Upload date:
  • Size: 9.4 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/7.0.0 CPython/3.13.14

File hashes

Hashes for trolleybus-0.2.0-py3-none-any.whl
Algorithm Hash digest
SHA256 36e014879cd0f6fbfadcb897a2ff7a46076508a81dd564e2a8d3961be4c05f91
MD5 402ace0f1425dedc79bb9f2651cef829
BLAKE2b-256 f118da66cf0353037386546ab99ef630dca6c9a0c5c4db9d3e7d007c9d2fc574

See more details on using hashes here.

Provenance

The following attestation bundles were made for trolleybus-0.2.0-py3-none-any.whl:

Publisher: publish.yml on blanebf/trolleybus

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Release history Release notifications | RSS feed

This release

0.2.0 This release

2 files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page