{"id":2421,"library":"bubus","title":"Bubus Event Bus","description":"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.","status":"active","version":"1.5.6","language":"en","source_language":"en","source_url":"https://github.com/browser-use/bubus","tags":["event-bus","async","pydantic","events","messaging","concurrency"],"install":[{"cmd":"pip install bubus","lang":"bash","label":"Install latest version"}],"dependencies":[{"reason":"Core dependency for event data models and validation.","package":"pydantic"}],"imports":[{"symbol":"EventBus","correct":"from bubus import EventBus"},{"symbol":"BaseEvent","correct":"from bubus import BaseEvent"}],"quickstart":{"code":"import asyncio\nfrom bubus import EventBus, BaseEvent\n\nclass MyEvent(BaseEvent):\n    message: str\n\nasync def my_handler(event: MyEvent) -> None:\n    print(f\"Received event: {event.message}\")\n    await asyncio.sleep(0.01) # Simulate async work\n\nasync def main():\n    bus = EventBus()\n    bus.subscribe(MyEvent, my_handler)\n\n    print(\"Publishing event...\")\n    await bus.publish(MyEvent(message=\"Hello from Bubus!\"))\n\n    # Wait for all published events to be handled\n    await bus.wait_for_all_handlers()\n    await bus.stop() # Gracefully shut down the event bus\n    print(\"Bus stopped.\")\n\nif __name__ == \"__main__\":\n    asyncio.run(main())","lang":"python","description":"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."},"warnings":[{"fix":"Update any environment variables or configuration files that use `BUBUS_LOG_LEVEL` to `BUBUS_LOGGING_LEVEL`.","message":"The environment variable for setting the logging level was renamed from `BUBUS_LOG_LEVEL` to `BUBUS_LOGGING_LEVEL`.","severity":"breaking","affected_versions":">=1.5.4"},{"fix":"Adjust any error handling or `try...except` blocks around event `publish` calls to anticipate the direct propagation of handler-raised exceptions.","message":"As of version 1.5.6, `bubus` now raises original exceptions directly from event handlers, rather than potentially wrapping or suppressing them in certain scenarios.","severity":"breaking","affected_versions":">=1.5.6"},{"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: ...`.","message":"To 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]`.","severity":"gotcha","affected_versions":">=1.5.0"},{"fix":"Always use the latest stable version of `bubus` when deploying in a multiprocessing context to leverage performance and stability improvements in semaphore management.","message":"When 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.","severity":"gotcha","affected_versions":"<1.4.7"}],"env_vars":null,"last_verified":"2026-04-10T00:00:00.000Z","next_check":"2026-07-09T00:00:00.000Z"}