abxbus

What it is and what it does

Abxbus is an in-memory event bus designed for building event-driven applications in async Python. You define events as Pydantic models, register handlers (sync or async), and emit events through a bus instance. The library handles automatic UUIDv7 generation, monotonic nanosecond timestamps, nested event tracking, and provides options for strict FIFO or fully parallel processing via built-in locking.

It's built to scale from single events to millions per second while keeping the API simple and consistent. The package includes features like event pattern matching (by class, string type, or wildcard), handler return value collection, event forwarding between separate bus instances with automatic loop prevention, and timeout enforcement across nested event hierarchies. It's inspired by JavaScript EventEmitter patterns but adds compile-time type safety through Pydantic and cross-runtime schema compatibility.

Use it for

• Build microservice architectures where different services communicate via typed events forwarded between separate EventBus instances.
• Implement complex async workflows with nested event handling and automatic cancellation when parent events time out.
• Create real-time applications that emit and handle thousands of events per second with predictable, measurable latency.
• Define strongly-typed event schemas that can be validated at runtime and potentially shared across Python and other runtimes.
• Decouple event producers from consumers in a single application using pattern-based subscriptions (by class, string, or wildcard).

Worth the install?

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

Install

abxbus on PyPI

Before you install

Low friction install with only three runtime dependencies (pydantic, typing-extensions, uuid7). Active maintenance with a recent release (13 days old) and steady commits; early project (first release March 2026) but no known vulnerabilities.

Requires Python 3.11 or later.

License in practice

MIT license (permissive) — you can use, modify, and distribute freely with minimal restrictions.

Quickstart

import asyncio
from abxbus import BaseEvent, EventBus

class SomeEvent(BaseEvent):
    some_data: int

async def handle_some_event(event: SomeEvent):
    print('hi!')

async def main():
    bus = EventBus()
    bus.on(SomeEvent, handle_some_event)
    await bus.emit(SomeEvent(some_data=132)).now()

asyncio.run(main())

Package facts

Evidence: abxbus-2.5.45-py3-none-any.whl

Tags

More Application Frameworks packages

See also bubus · pyventus · abx-dl · pyee · azure-servicebus · aws-cdk.aws-events · rel · observable · abxpkg · pyserial-asyncio