bubus

What it is and what it does

Bubus is an event bus library that lets you build event-driven applications in Python using Pydantic models for type-safe event definitions. It handles both async and sync handlers, maintains strict FIFO processing order, and supports nested event dispatch—so handlers can emit child events that are tracked as part of the causality tree. You define events as Pydantic BaseEvent subclasses, register handlers with bus.on(), and dispatch events with bus.dispatch(). The library manages concurrency, result aggregation from multiple handlers, and can forward events between separate bus instances (useful for microservice-like architectures) while preventing infinite loops.

The main dependencies are pydantic (for validation), anyio and aiofiles (for async I/O), portalocker (for file-based locking), and uuid7 (for event IDs). It's designed to feel familiar to developers coming from JavaScript's EventTarget API, bringing that simplicity to Python with full type checking. The library is actively maintained, has no known vulnerabilities, and sits in the top 5000 packages by download volume.

Use it for

• Build microservice architectures where different services communicate via event buses that forward events between each other.
• Implement real-time notification systems where multiple handlers subscribe to the same event and aggregate results.
• Create async workflows where handlers dispatch child events and wait for completion, maintaining a causality tree.
• Design type-safe domain event systems where Pydantic validation ensures all events conform to their schema.
• Handle concurrent async operations with guaranteed FIFO ordering, even when handlers spawn nested events.

Worth the install?

AI-flagged interpretation of the facts on this page. Verify before relying on it.

Install

bubus on PyPI

Before you install

Installation is low-friction—a pure Python wheel with six runtime dependencies (aiofiles, anyio, portalocker, pydantic, typing-extensions, uuid7). The package is actively maintained with a recent commit history and has been in active development since June 2025.

Requires Python 3.11 or later; all handlers and event dispatch are async-first, so you must run within an async context.

License in practice

Licensed under MIT (permissive), so you can use, modify, and distribute it freely in commercial and open-source projects without restriction.

Quickstart

pip install bubus

from bubus import EventBus, BaseEvent

class UserLoginEvent(BaseEvent):
    username: str

async def handle_login(event: UserLoginEvent) -> str:
    return f"User {event.username} logged in"

bus = EventBus()
bus.on(UserLoginEvent, handle_login)
event = bus.dispatch(UserLoginEvent(username="alice"))
result = await event.event_result()

Package facts

Evidence: bubus-1.5.6-py3-none-any.whl

Tags

More Application Frameworks packages

See also abxbus · pyventus · azure-servicebus · observable · fastapi-events · aws-cdk.aws-events · queuelib · ovos-bus-client · Pypubsub · PyDispatcher