bubus-py310x

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())