InfinityBus — Async Event Bus

Strongly-typed, extensible async event bus — middleware pipeline + advanced templates.

📖 中文版

📑 Table of Contents

• ✨ Features
• 🔍 Comparison
• 📦 Installation
• 🚀 Quick Start
• 🧱 Architecture
• 📚 Documentation
• 🧪 Testing
• 🧑‍💻 Development
• 📄 License

✨ Features

Unlike similar projects: InfinityBus is extensible (middleware onion pipeline), rather than hardcoding all functionality in the core class. See Middleware System for details.

🔍 Comparison

Choose InfinityBus when: you need a middleware-native architecture — hot-reloadable onion pipeline with 8 built-in middlewares (logging, metrics, rate-limit, transform, block, recursion guard, forwarding); you run pyright strict across the entire codebase with literally zero # type: ignore in any production code; you want full production-grade features (backpressure, graceful shutdown, regex subscriptions, Pydantic payload validation) with a single core dependency.

Choose bubus when: you need built-in event return values and event forwarding; you need Python 3.11 support; you prefer the class-driven API style of bus.on(EventClass, fn).

Choose pyee when: you're familiar with the Node.js EventEmitter style; you need both asyncio and trio support; you prefer a minimal API (ee.on('event', fn)); the type system has been greatly enhanced since v13, passing both mypy and pyright.

Choose PyPubSub when: you don't use asyncio; you need very broad Python version compatibility (3.7–3.14); you prefer traditional topic hierarchy string matching.

Full commits: · InfinityBus d7a67f0 · bubus 7c09342 · pyee 5157de2 (no coverage in official CI; manually measured via pytest-cov) · PyPubSub 4ec2c47

📦 Installation

We recommend uv — a blazing-fast Python package manager, 10–100× faster than pip, with automatic venv management, dependency locking, and conflict resolution. No need for separate virtual environment tools.

pip

# Core (pub/sub and middleware pipeline only)
pip install infinity_bus

# Core + advanced templates (expect, request, pipe, SQLite logging, etc.)
pip install infinity_bus[templates]

uv (recommended)

# Core
uv add infinity_bus

# Core + advanced templates
uv add infinity_bus --extra templates

Note: Advanced templates (expect, request, pipe, SQLiteLoggingMiddleware) require the [templates] extra. Using them without installing the extra will raise an ImportError.

Or install from source:

git clone https://github.com/yinbailiang/event_bus.git
cd event_bus
uv sync --extra dev

🚀 Quick Start

Basic Pub/Sub

Using the handler simplified API requires: pip install infinity_bus[templates] Or with uv: uv add infinity_bus --extra templates

import asyncio
from pydantic import BaseModel
from event_bus import (
    EventBus, EventDeclaration,
    EventRegistry, EventHandlerRegistry,
)
from event_bus.templates import handler

# 1. Define payload
class MyPayload(BaseModel):
    message: str

# 2. Declare event
class MyEvent(EventDeclaration):
    name = "my.event"
    payload_type = MyPayload

# 3. Implement handler (simplified: function + decorator)
@handler(MyEvent)
async def my_handler(payload: MyPayload) -> None:
    print(f"Received: {payload.message}")

# 4. Wire up and run
async def main():
    reg = EventRegistry()
    reg.register(MyEvent)
    h_reg = EventHandlerRegistry()
    h_reg.register(my_handler())

    async with EventBus(reg, h_reg) as bus:
        await bus.proxy("cli").publish("my.event", {"message": "Hello, EventBus!"})
        await asyncio.sleep(1)  # wait for handler output

asyncio.run(main())

Request-Response Pattern

Using advanced templates like request / expect / pipe requires: pip install infinity_bus[templates] Or with uv: uv add infinity_bus --extra templates

import asyncio
from pydantic import BaseModel
from event_bus import (
    EventBus, EventDeclaration, EventHandler,
    EventRegistry, EventHandlerRegistry,
)
from event_bus.templates.request import (
    request, RequestProtocol, ResponseProtocol,
)

# 1. Define request/response payloads
class GetUserRequest(RequestProtocol):
    user_id: int

class GetUserResponse(ResponseProtocol):
    user_name: str
    email: str

# 2. Declare events
class GetUserRequestEvent(EventDeclaration):
    name = "user.get.request"
    payload_type = GetUserRequest

class GetUserResponseEvent(EventDeclaration):
    name = "user.get.response"
    payload_type = GetUserResponse

# 3. Implement server-side handler
class GetUserHandler(EventHandler):
    def __init__(self):
        super().__init__(subscriptions=["user.get.request"])

    async def handle(self, payload, bus_proxy, raw_event):
        if not isinstance(payload, GetUserRequest):
            return
        resp = GetUserResponse(
            session_id=payload.session_id,
            request_id=payload.request_id,
            success=True,
            user_name="Alice",
            email="alice@example.com",
        )
        await bus_proxy.publish("user.get.response", resp)

# 4. Wire up and run
async def main():
    reg = EventRegistry()
    reg.register(GetUserRequestEvent)
    reg.register(GetUserResponseEvent)
    h_reg = EventHandlerRegistry()
    h_reg.register(GetUserHandler())

    async with EventBus(reg, h_reg) as bus:
        proxy = bus.proxy("cli")
        resp = await request(
            bus_proxy=proxy,
            req_event="user.get.request",
            req_data={"user_id": 123},
            resp_event="user.get.response",
            timeout=10.0,
        )
        resp.raise_if_failed()
        print(f"User: {resp.user_name} ({resp.email})")

asyncio.run(main())

🧱 Architecture

📚 Documentation

🧪 Testing

pip

# Clone and install test dependencies
git clone https://github.com/yinbailiang/event_bus.git
cd event_bus
pip install -e ".[test]"

# Run all tests
pytest --cov=src -v

# Run template tests only
pytest tests/templates/ -v

uv

# Clone and sync test dependencies
git clone https://github.com/yinbailiang/event_bus.git
cd event_bus
uv sync --extra test

# Run all tests
uv run pytest --cov=src -v

# Run template tests only
uv run pytest tests/templates/ -v

🧑‍💻 Development

The sole toolchain for developing this project is uv. The pre-commit gating invokes tests and code quality tools via uv run — please install uv first.

# 1. Clone and sync all dependencies (auto-creates venv)
git clone https://github.com/yinbailiang/event_bus.git
cd event_bus
uv sync --extra dev

# 2. Install pre-commit hooks
uv run pre-commit install

Development Loop

# lint + format
uv run ruff check src/ && uv run ruff format src/ --check

# type check
uv run pyright src/

# test + coverage
uv run pytest --cov=src -v

# run all gates manually
uv run pre-commit run --all-files

📖 See CONTRIBUTING.md for the full guide.

📄 License

MIT

Part of InfinitySystem