Pre-release

This release is a pre-release and may not be stable for production use.

Buz

Buz is a lightweight, simple, and extensible Python library that provides implementations of Event, Command, and Query buses following CQRS and Event-Driven Architecture patterns.

📋 Table of Contents

✨ Key Features

🚌 Bus Types: Event, Command, and Query buses for clean architecture
🔄 Sync & Async Support: Both synchronous and asynchronous implementations
🔧 Middleware System: Extensible middleware for cross-cutting concerns
📦 Message Brokers: Support for Kafka, RabbitMQ (via Kombu), and in-memory
🔒 Transactional Outbox: Reliable event publishing with transactional guarantees
🎯 Dependency Injection: Built-in locator pattern for handler resolution
📝 Type Safety: Fully typed with mypy support
🪶 Lightweight: Minimal dependencies, maximum flexibility

🚀 Quick Start

Installation

# Basic installation
pip install buz

# With Kafka support
pip install buz[aiokafka]

# With RabbitMQ support
pip install buz[kombu]

# With dependency injection
pip install buz[pypendency]

Basic Usage

Event Bus Example

from dataclasses import dataclass
from buz import Message
from buz.event import Event, BaseSubscriber
from buz.event.sync import SyncEventBus
from buz.locator.sync import InstanceLocator

@dataclass(frozen=True)
class UserCreated(Event):
    user_id: str
    email: str

class EmailSubscriber(BaseSubscriber):
    def consume(self, event: UserCreated) -> None:
        print(f"Sending welcome email to {event.email}")

class AnalyticsSubscriber(BaseSubscriber):
    def consume(self, event: UserCreated) -> None:
        print(f"Tracking user creation: {event.user_id}")

# Setup
locator: InstanceLocator = InstanceLocator()
locator.register(EmailSubscriber())
locator.register(AnalyticsSubscriber())

event_bus = SyncEventBus(locator)

# Usage
event = UserCreated(user_id="123", email="user@example.com")
event_bus.publish(event)

Command Bus Example

from dataclasses import dataclass
from buz.command import Command
from buz.command.synchronous import BaseCommandHandler
from buz.command.synchronous.self_process import SelfProcessCommandBus
from buz.locator.sync import InstanceLocator

@dataclass(frozen=True)
class CreateUser(Command):
    email: str
    name: str

class CreateUserCommandHandler(BaseCommandHandler):
    def handle(self, command: CreateUser) -> None:
        # Business logic here
        print(f"Creating user: {command.name} ({command.email})")

# Setup
locator = InstanceLocator()
locator.register(CreateUserCommandHandler())

command_bus = SelfProcessCommandBus(locator)

# Usage
command = CreateUser(email="user@example.com", name="John Doe")
command_bus.handle(command)

Query Bus Example

from dataclasses import dataclass
from buz.query import Query, QueryResponse
from buz.query.synchronous import BaseQueryHandler
from buz.query.synchronous.self_process import SelfProcessQueryBus
from buz.locator.sync import InstanceLocator

@dataclass(frozen=True)
class GetUser(Query):
    user_id: str

@dataclass(frozen=True)
class User:
    user_id: str
    name: str
    email: str

class GetUserQueryHandler(BaseQueryHandler):
    def handle(self, query: GetUser) -> QueryResponse:
        # Business logic here
        return QueryResponse(
            content=User(
                user_id=query.user_id,
                name="John Doe",
                email="john@example.com"
            )
        )

# Setup
locator = InstanceLocator()
locator.register(GetUserQueryHandler())

query_bus = SelfProcessQueryBus(locator)

# Usage
query = GetUser(user_id="123")
query_response = query_bus.handle(query)
user = query_response.content
print(f"User: {user.name}")

🏗️ Architecture

Buz implements the Command Query Responsibility Segregation (CQRS) pattern with distinct buses:

Event Bus

Purpose: Publish domain events and notify multiple subscribers
Pattern: Pub/Sub with multiple handlers per event
Use Cases: Domain event broadcasting, eventual consistency, integration events

Command Bus

Purpose: Execute business operations and commands
Pattern: Single handler per command
Use Cases: Business logic execution, write operations, state changes

Query Bus

Purpose: Retrieve data and execute queries
Pattern: Single handler per query with typed responses
Use Cases: Data retrieval, read operations, projections

🔧 Advanced Features

Middleware System

Add cross-cutting concerns like logging, validation, and metrics:

from datetime import datetime
from buz.event import Event, Subscriber
from buz.event.middleware import BasePublishMiddleware, BaseConsumeMiddleware
from buz.event.infrastructure.models.execution_context import ExecutionContext

class LoggingPublishMiddleware(BasePublishMiddleware):
    def _before_on_publish(self, event: Event) -> None:
        print(f"Publishing event {event}")

    def _after_on_publish(self, event: Event) -> None:
        return

class MetricsConsumeMiddleware(BaseConsumeMiddleware):
    def __init__(self) -> None:
        self.__consumption_start_time: datetime = datetime.now()

    def _before_on_consume(
        self,
        event: Event,
        subscriber: Subscriber,
        execution_context: ExecutionContext,
    ) -> None:
        self.__consumption_start_time = datetime.now()

    def _after_on_consume(
        self,
        event: Event,
        subscriber: Subscriber,
        execution_context: ExecutionContext,
    ) -> None:
        consumption_time_ms = int((datetime.now() - self.__consumption_start_time).total_seconds() * 1000)
        print(
            f"Subscriber {subscriber.fqn()} consumed event {event.id} successfully in {consumption_time_ms} ms"
        )

# Apply middleware
event_bus = SyncEventBus(
    locator=locator,
    publish_middlewares=[LoggingPublishMiddleware()],
    consume_middlewares=[MetricsConsumeMiddleware()]
)

# Usage
event = UserCreated(user_id="123", email="user@example.com")
event_bus.publish(event)

Transactional Outbox Pattern

Ensure reliable event publishing with database transactions:

from buz.event.transactional_outbox import TransactionalOutboxEventBus

# Configure with your database and event bus
transactional_outbox_bus = TransactionalOutboxEventBus(
    outbox_repository=your_outbox_repository,
    event_to_outbox_record_translator=your_outbox_record_translator,
    ...
)

# Events are stored in database, published later by worker
transactional_outbox_bus.publish(event)

RabbitMQ

from buz.event.infrastructure.kombu.kombu_event_bus import KombuEventBus

kombu_event_bus = KombuEventBus(
    connection=your_connection,
    publish_strategy=your_publish_strategy,
    publish_retry_policy=you_publish_retry_policy,
    ...
)

# Published and consumed in RabbitMQ
kombu_event_bus.publish(event)

Kafka Integration

from buz.kafka import BuzKafkaEventBus

kafka_bus = KafkaEventBus(
    publish_strategy=your_publish_strategy,
    producer=your_producer,
    logger=your_logger,
    ...
)

# Published and consumed in Kafka
kafka_bus.publish(event)

Async Support

from buz.event.async_event_bus import AsyncEventBus
from buz.query.asynchronous import QueryBus as AsyncQueryBus
from buz.command.asynchronous import CommandHandler as AsyncCommandHandler


# Async event bus
async_event_bus = AsyncEventBus(locator)
await async_event_bus.publish(event)

# Async query bus
async_query_bus = AsyncQueryBus(locator)
await async_query_bus.handle(event)

# Async command bus
async_command_bus = AsyncCommandBus(locator)
await async_command_bus.handle(command)

📦 Message Brokers

Supported Brokers

Broker	Sync	Async	Installation
In-Memory	✅	✅	Built-in
Kafka	✅	✅	`pip install buz[aiokafka]`
RabbitMQ	✅	❌	`pip install buz[kombu]`

🧪 Testing

Buz includes testing utilities for unit and integration tests:

from buz.event.sync import SyncEventBus
from buz.locator.sync import InstanceLocator

test_locator = InstanceLocator()
test_bus = SyncEventBus(test_locator)

test_locator.register(EmailSubscriber())
test_bus.publish(UserCreated(user_id="123", email="test@example.com"))

🔗 Related Projects

buz-fever-shared: Opinionated utilities and standards for Buz
buz-basic-example: Complete example project with Docker setup

📋 Requirements

Python 3.9+
Optional dependencies based on features used

🤝 Contributing

We welcome contributions! Please see our Contributing Guidelines for details.

Development Setup

# Clone the repository
git clone https://github.com/Feverup/buz.git
cd buz

# Install with development dependencies
make build

# Run tests
make test

# Run linting
make lint

# Format code
make format

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

📚 Documentation

Changelog - Release notes and version history

🙋‍♀️ Support

Create an Issue for bug reports or feature requests

Made with ❤️ by the Fever Platform Team

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

buz-6.0.0rc1.tar.gz (73.0 kB view details)

Uploaded Aug 26, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

buz-6.0.0rc1-py3-none-any.whl (173.6 kB view details)

Uploaded Aug 26, 2026 Python 3

File details

Details for the file buz-6.0.0rc1.tar.gz.

File metadata

Download URL: buz-6.0.0rc1.tar.gz
Upload date: Aug 26, 2026
Size: 73.0 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: poetry/1.8.3 CPython/3.10.21 Linux/6.17.0-1022-azure

File hashes

Hashes for buz-6.0.0rc1.tar.gz
Algorithm	Hash digest
SHA256	`8b6cb39a4cb5b86658b18f42df6d12acf29126d03aea1feb47c09ea91015151f`
MD5	`2bc20c29efce54d9cca3e68f48951c77`
BLAKE2b-256	`6d2f27fec9c71d4a30c29005f1db513686e70ea061e92da55037597d9db9f3ff`

See more details on using hashes here.

File details

Details for the file buz-6.0.0rc1-py3-none-any.whl.

File metadata

Download URL: buz-6.0.0rc1-py3-none-any.whl
Upload date: Aug 26, 2026
Size: 173.6 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: poetry/1.8.3 CPython/3.10.21 Linux/6.17.0-1022-azure

File hashes

Hashes for buz-6.0.0rc1-py3-none-any.whl
Algorithm	Hash digest
SHA256	`c31b9fcd0063fdd3b324d6ad31f1641b05c70b6a6e6b53fa0549df7d35a97a02`
MD5	`14e1d6d44be575fd6066e9803c0d8807`
BLAKE2b-256	`325f4f6e0ab6665d00996f7c3edb9b1a1756caae790a2df4d0b89d59688deb49`