{"id":7058,"library":"bubus-py310x","title":"Bubus (Python 3.10+ Fork)","description":"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.","status":"active","version":"1.3.0","language":"en","source_language":"en","source_url":"https://github.com/abhijeetkaze/bubus","tags":["event bus","async","pydantic","python3.10","fork"],"install":[{"cmd":"pip install bubus-py310x","lang":"bash","label":"Install `bubus-py310x`"}],"dependencies":[{"reason":"Core functionality relies on Pydantic for event model definition 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\nfrom pydantic import BaseModel\n\nclass UserLoginEvent(BaseEvent[str]):\n    username: str\n    is_admin: bool\n\n# Dummy for demonstration, replace with actual API call if needed\nclass AuthRequestEvent(BaseEvent[None]):\n    pass\n\nclass AuthResponseEvent(BaseEvent[str]):\n    message: str\n\nclass AuthAPI:\n    @staticmethod\n    async def post(event: AuthRequestEvent):\n        print(\"AuthAPI received AuthRequestEvent. Dispatching AuthResponseEvent.\")\n        await asyncio.sleep(0.1) # Simulate async work\n        event.event_bus.dispatch(AuthResponseEvent(message=\"Auth successful!\"))\n\n\nasync def handle_login(event: UserLoginEvent) -> str:\n    print(f\"Handler for UserLoginEvent received: {event.username}, Admin: {event.is_admin}\")\n    auth_request = await event.event_bus.dispatch(AuthRequestEvent()) # nested events supported\n    auth_response = await event.event_bus.expect(AuthResponseEvent, timeout=30.0) # Uses 3.10 compatible timeout\n    return f\"User {event.username} logged in admin={event.is_admin} with API response: {auth_response.message}\"\n\nasync def main():\n    bus = EventBus()\n    bus.on(UserLoginEvent, handle_login)\n    bus.on(AuthRequestEvent, AuthAPI.post)\n\n    print(\"Dispatching UserLoginEvent...\")\n    event_result = await bus.dispatch(UserLoginEvent(username=\"alice\", is_admin=True)).event_result()\n    print(f\"Final result: {event_result}\")\n\nif __name__ == \"__main__\":\n    asyncio.run(main())","lang":"python","description":"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."},"warnings":[{"fix":"Use `bubus-py310x` if targeting Python 3.10 to ensure `asyncio.timeout` compatibility.","message":"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.","severity":"gotcha","affected_versions":"Original `bubus` on Python 3.10, `bubus-py310x` (solves this)"},{"fix":"Ensure your event handlers conform to the specified `BaseEvent` return type or omit the generic parameter if no strict return type is required.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions"}],"env_vars":null,"last_verified":"2026-04-16T00:00:00.000Z","next_check":"2026-07-15T00:00:00.000Z","problems":[{"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').","cause":"Attempting to import from 'bubus' when 'bubus-py310x' is installed, or `bubus-py310x` is not installed.","error":"ModuleNotFoundError: No module named 'bubus'"},{"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`.","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.","error":"TypeError: object NoneType can't be used in 'await' expression"},{"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`).","cause":"The event data being dispatched does not conform to the Pydantic model definition of the `BaseEvent` subclass.","error":"pydantic.error_wrappers.ValidationError: ..."}]}