Bubus (Python 3.10+ Fork)

1.3.0 · active · verified Thu Apr 16

Bubus-py310x is a modified fork of the `bubus` event bus library, specifically designed to support Python 3.10 by replacing usages of `asyncio.timeout` (introduced in Python 3.11) with compatible alternatives. It provides a Pydantic-powered, fully-featured event bus for building asynchronous Python applications, supporting type-safe publish-subscribe patterns with async/sync handler support and advanced features like event forwarding and concurrency control. The current version is 1.3.0, and it maintains an active release cadence to ensure Python 3.10 compatibility.

Common errors

```
ModuleNotFoundError: No module named 'bubus'
```
cause Attempting to import from 'bubus' when 'bubus-py310x' is installed, or `bubus-py310x` is not installed.

fix Ensure you have installed `bubus-py310x` using `pip install bubus-py310x` and your import statements are `from bubus import ...` (the module name remains 'bubus').
```
TypeError: object NoneType can't be used in 'await' expression
```
cause This can occur if an `await` expression is used on a function that does not return an awaitable (e.g., an `async` function not being `await`ed, or a `dispatch` call not properly yielding). Specific to Python 3.10, this can also happen if attempting to use the original `bubus` library's `asyncio.timeout` calls, which are not compatible.

fix Ensure all `async` functions are `await`ed. If using `bubus-py310x` for Python 3.10, this specific `asyncio.timeout` issue is resolved by the fork. If still encountering, verify your handlers and event dispatches are correctly structured for `async`/`await`.
```
pydantic.error_wrappers.ValidationError: ...
```
cause The event data being dispatched does not conform to the Pydantic model definition of the `BaseEvent` subclass.

fix Review the data passed when creating an event instance (e.g., `UserLoginEvent(username='alice', is_admin=True)`) and ensure it matches the fields and types defined in your `BaseEvent` subclass (e.g., `UserLoginEvent`).

Warnings

gotcha If migrating from the original `bubus` library to `bubus-py310x` on Python 3.10, be aware that the original library's reliance on `asyncio.timeout` (Python 3.11+) caused compatibility issues. `bubus-py310x` explicitly addresses this by replacing those usages with Python 3.10 compatible alternatives.
Fix: Use `bubus-py310x` if targeting Python 3.10 to ensure `asyncio.timeout` compatibility.
gotcha Bubus supports strict typing for event handler return values using `BaseEvent[ReturnTypeHere]`. If the generic parameter `ReturnTypeHere` is provided, `bubus` will enforce that all registered handlers for that event must return `ReturnTypeHere | None` at compile-time (via type hints) and at runtime.
Fix: Ensure your event handlers conform to the specified `BaseEvent` return type or omit the generic parameter if no strict return type is required.
gotcha EventBus instances can accumulate events in memory if not properly managed, potentially leading to memory leaks, especially with large or long-lived events. The library provides `max_history_size` for automatic cleanup and `bus.stop(clear=True)` to free memory for unused buses.
Fix: Configure `max_history_size` on your `EventBus` for automatic cleanup, or explicitly call `bus.stop(clear=True)` when an `EventBus` instance is no longer needed.

Install

pip install bubus-py310x Install `bubus-py310x`

Imports

EventBus
```
from bubus import EventBus
```
BaseEvent
```
from bubus import BaseEvent
```

Quickstart

This quickstart demonstrates how to set up an `EventBus`, define `BaseEvent` models with Pydantic, register asynchronous handlers, and dispatch/expect events. It showcases nested event handling and the use of `bus.expect` with a timeout, which is made compatible with Python 3.10 by this fork.

import asyncio
from bubus import EventBus, BaseEvent
from pydantic import BaseModel

class UserLoginEvent(BaseEvent[str]):
    username: str
    is_admin: bool

# Dummy for demonstration, replace with actual API call if needed
class AuthRequestEvent(BaseEvent[None]):
    pass

class AuthResponseEvent(BaseEvent[str]):
    message: str

class AuthAPI:
    @staticmethod
    async def post(event: AuthRequestEvent):
        print("AuthAPI received AuthRequestEvent. Dispatching AuthResponseEvent.")
        await asyncio.sleep(0.1) # Simulate async work
        event.event_bus.dispatch(AuthResponseEvent(message="Auth successful!"))


async def handle_login(event: UserLoginEvent) -> str:
    print(f"Handler for UserLoginEvent received: {event.username}, Admin: {event.is_admin}")
    auth_request = await event.event_bus.dispatch(AuthRequestEvent()) # nested events supported
    auth_response = await event.event_bus.expect(AuthResponseEvent, timeout=30.0) # Uses 3.10 compatible timeout
    return f"User {event.username} logged in admin={event.is_admin} with API response: {auth_response.message}"

async def main():
    bus = EventBus()
    bus.on(UserLoginEvent, handle_login)
    bus.on(AuthRequestEvent, AuthAPI.post)

    print("Dispatching UserLoginEvent...")
    event_result = await bus.dispatch(UserLoginEvent(username="alice", is_admin=True)).event_result()
    print(f"Final result: {event_result}")

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

view raw JSON →