bubus

library1.5.6pypypi✓ verified 25d ago

Bubus is an advanced, Pydantic-powered asynchronous event bus library for Python. It facilitates decoupled communication between components by allowing them to publish and subscribe to events. It's designed for modern Python applications, emphasizing type safety, performance, and robust error handling. The current version is 1.5.6, and it receives regular updates.

workflow web-framework serialization

pip install bubus

github ↗package ↗

INSTALL

IMPORT

SIG · BUBUS

bubus

workflowpythonv1.5.6

Install

3.3s avg

Import

778ms

Disk

28MB

Pass rate

6/ 10

Env Coverage6 / 10

glibc

3.9–3.13

musl

3.9–3.13

Install & Compatibility

Where this runs

tested against v1.5.6 · pip install

no network on importno background threads

Install × environment matrix

Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux

musl

glibc

py 3.10

✕ build_error

py 3.11

✓ —

✓ 3.7s

py 3.12

✓ —

✓ 3s

py 3.13

✓ —

✓ 3.3s

py 3.9

✕ build_error

28MB installed

● package 28MB

Code

Verified usage

Verified import paths — ran on the pinned version, not inferred.

EventBus

✓ from bubus import EventBus

BaseEvent

✓ from bubus import BaseEvent

This quickstart demonstrates how to set up an `EventBus`, define a `BaseEvent` with Pydantic, subscribe an asynchronous handler to it, and publish an event. It then shows how to wait for all handlers to complete and gracefully stop the bus.

import asyncio from bubus import EventBus, BaseEvent class MyEvent(BaseEvent): message: str async def my_handler(event: MyEvent) -> None: print(f"Received event: {event.message}") await asyncio.sleep(0.01) # Simulate async work async def main(): bus = EventBus() bus.subscribe(MyEvent, my_handler) print("Publishing event...") await bus.publish(MyEvent(message="Hello from Bubus!")) # Wait for all published events to be handled await bus.wait_for_all_handlers() await bus.stop() # Gracefully shut down the event bus print("Bus stopped.") if __name__ == "__main__": asyncio.run(main())

Debug

Known issues

breakingThe environment variable for setting the logging level was renamed from `BUBUS_LOG_LEVEL` to `BUBUS_LOGGING_LEVEL`.

fix

Update any environment variables or configuration files that use `BUBUS_LOG_LEVEL` to `BUBUS_LOGGING_LEVEL`.

affects: >=1.5.4

breakingAs of version 1.5.6, `bubus` now raises original exceptions directly from event handlers, rather than potentially wrapping or suppressing them in certain scenarios.

fix

Adjust any error handling or `try...except` blocks around event `publish` calls to anticipate the direct propagation of handler-raised exceptions.

affects: >=1.5.6

gotchaTo enforce return types for event handlers and utilize the new return type enforcement feature, `BaseEvent` should be defined as a Generic, e.g., `BaseEvent[ReturnType]`.

fix

If you require return type validation, define your event classes like `class MyEvent(BaseEvent[str]): ...` and ensure handlers match `async def handler(event: MyEvent) -> str: ...`.

affects: >=1.5.0

gotchaWhen using `bubus` in multiprocessing environments, ensure you are on a recent version (>=1.4.7) to benefit from fixes related to semaphore handling and avoid potential issues with stale event loops or deleted semaphores.

fix

Always use the latest stable version of `bubus` when deploying in a multiprocessing context to leverage performance and stability improvements in semaphore management.

affects: <1.4.7

Errors

Common errors & fixes

ModuleNotFoundError: No module named 'bubus.bus'

Users often attempt to import the main event bus class from a non-existent submodule like 'bubus.bus' instead of directly from the top-level 'bubus' package.

fix

Correct the import statement to `from bubus import BubusBus` (assuming 'BubusBus' is the primary event bus class).

pydantic_core._pydantic_core.ValidationError: 1 validation error for MyEvent field_name Input should be a valid string [type=string_type, input_value=123, input_type=int]

The data provided when creating or publishing an event, which is internally a Pydantic model, does not conform to the event model's defined schema (e.g., incorrect data type, missing required fields).

fix

Ensure that the data (dictionary or keyword arguments) passed to the event constructor or during event creation strictly matches the Pydantic model's field types and constraints.

RuntimeWarning: coroutine 'BubusBus.publish' was never awaited

An asynchronous method of the `BubusBus` (like `publish` or `subscribe`) was called without the `await` keyword within an `async` function, or from a synchronous context without `asyncio.run()` to manage its execution.

fix

If calling from an `async` function, prepend `await` to the method call (e.g., `await bus.publish(event)`). If calling from a synchronous context, use `asyncio.run(bus.publish(event))` to execute the coroutine.

TypeError: Event handler for <EventType> must be an awaitable function.

An event handler function was registered with the `bubus` event bus using a standard synchronous `def` definition, but `bubus` requires all event handlers to be asynchronous functions (`async def`) due to its async nature.

fix

Modify the event handler function definition from `def my_handler(...)` to `async def my_handler(...)` to make it awaitable.

Upgrade

Version history

1.5.6latest on PyPI · released Aug 30, 2025

Audit

Dependencies

pydanticrequiredCore dependency for event data models and validation.

Agent activity

23 hits · last 30 days

node

OpenAI (training)

Resources

githubhttps://github.com/browser-use/bubus ↗

packagehttps://pypi.org/project/bubus/ ↗