Install
- Python
- TypeScript
First event
- Python
- TypeScript
Next steps
- Browse the Features section for behavior patterns.
- Use API Reference for signatures and options.
- See Integrations for bridges and middleware.
Get started quickly with bubus in Python or TypeScript.
pip install bubus
npm install bubus
import asyncio
from bubus import BaseEvent, EventBus
class CreateUserEvent(BaseEvent[dict]):
email: str
async def on_create_user(event: CreateUserEvent) -> dict:
user = await your_create_user_logic(event.email)
return {'user_id': user['id']}
async def main() -> None:
bus = EventBus('MyAuthEventBus')
bus.on(CreateUserEvent, on_create_user)
result = await bus.emit(CreateUserEvent(email='[email protected]')).event_result()
print(result)
# {'user_id': 'some-user-uuid'}
asyncio.run(main())
import { BaseEvent, EventBus } from 'bubus'
import { z } from 'zod'
const CreateUserEvent = BaseEvent.extend('CreateUserEvent', {
email: z.string(),
event_result_type: z.object({ user_id: z.string() }),
})
const bus = new EventBus('MyAuthEventBus')
bus.on(CreateUserEvent, async (event) => {
const user = await yourCreateUserLogic(event.email)
return { user_id: user.id }
})
const event = bus.emit(CreateUserEvent({ email: '[email protected]' }))
await event.done()
console.log(event.event_result) // { user_id: 'some-user-uuid' }