Protobus Python

A lightweight, scalable microservices message bus for Python. Leverages RabbitMQ for message routing and load balancing, combined with Protocol Buffers for efficient, type-safe serialization.

Note: This is the official Python port of Protobus, originally written in TypeScript. The API and architecture are designed to be as close to the original as possible.

Why Protobus?

Unlike transport-agnostic frameworks that abstract away the message broker, Protobus embraces RabbitMQ's native capabilities directly. We leverage topic exchanges, routing keys, competing consumers, dead-letter queues, and message persistence - rather than re-implementing routing logic at the application level.

RabbitMQ-Native Approach

Message Routing: By delegating routing to RabbitMQ's Erlang runtime instead of your Python process, Protobus eliminates the double-processing that transport-agnostic frameworks impose. The broker handles competing consumers, topic-based routing, and dead-letter queues natively.

Binary Serialization: Protocol Buffers provide 3-10x smaller payloads than JSON, with compile-time type safety and built-in backward compatibility. No more runtime schema guessing or JSON parsing overhead.

Polyglot Advantage

Since Protobus uses standard Protobuf schemas and AMQP protocol, implementing clients in other languages is straightforward. Currently available in TypeScript/Node.js and Python, with Java, Go, or Rust implementations requiring minimal effort.

Features

• RPC Communication: Request-response pattern over message queues
• Server Streaming: rpc foo (Req) returns (stream Chunk) — multi-chunk responses with async for ergonomics. See streaming docs.
• Event System: Publish-subscribe with topic-based routing and wildcards
• Auto-Reconnection: Exponential backoff with jitter for resilient connections
• Message Retry: Automatic retry with dead-letter queue (DLQ) support
• Message Priority: Opt-in priority queues so control traffic jumps a bulk backlog on the same queue. See priority docs — enabling it on an existing queue needs a one-time operator migration.
• Custom Types: Extensible type system (BigInt, Timestamp built-in)
• Async/Await: Built on asyncio and aio-pika for modern Python
• CLI Tools: Generate types and service stubs from .proto files
• Lifecycle Management: RunnableService with graceful shutdown handling

Requirements

• Python 3.10 or higher
• RabbitMQ 3.8 or higher

Installation

pip install protobus

Or install from source:

git clone https://github.com/ArielLaub/protobus-py.git
cd protobus-py
pip install -e .

Quick Start

1. Start RabbitMQ

docker-compose up -d

2. Create a Service

from protobus import RunnableService, Context

class CalculatorService(RunnableService):
    @property
    def service_name(self) -> str:
        return "calculator.MathService"

    async def add(self, data: dict, actor: str, correlation_id: str) -> dict:
        return {"result": data["a"] + data["b"]}

    async def multiply(self, data: dict, actor: str, correlation_id: str) -> dict:
        return {"result": data["a"] * data["b"]}

3. Run the Service

import asyncio
from protobus import Context

async def main():
    ctx = Context()
    await ctx.init("amqp://guest:guest@localhost:5672/")

    # Start with lifecycle management (handles SIGINT/SIGTERM)
    await CalculatorService.start(ctx, CalculatorService)

asyncio.run(main())

4. Create a Client

import asyncio
from protobus import Context, ServiceProxy

async def main():
    ctx = Context()
    await ctx.init("amqp://guest:guest@localhost:5672/")

    # Create proxy for the calculator service
    calc = ServiceProxy(ctx, "calculator.MathService")
    await calc.init()

    # Make RPC calls
    result = await calc.add({"a": 5, "b": 3})
    print(f"5 + 3 = {result['result']}")  # Output: 5 + 3 = 8

    result = await calc.multiply({"a": 4, "b": 7})
    print(f"4 * 7 = {result['result']}")  # Output: 4 * 7 = 28

    await ctx.close()

asyncio.run(main())

CLI Tools

Protobus includes CLI tools for code generation:

# Generate Python types from .proto files
protobus generate

# Generate a service stub
protobus generate:service calculator.MathService

# Show project setup instructions
protobus init

Configure in pyproject.toml:

[tool.protobus]
protoDir = "./proto"
typesOutput = "./types/proto.py"
servicesDir = "./services"

Documentation

• Getting Started - Installation and first service
• Architecture - System design and components
• Configuration - Environment variables and options
• Message Flow - How messages move through the system
• CLI Tools - Code generation commands
• Troubleshooting - Common issues and solutions

API Reference

• Context - Connection management
• MessageService - Building services
• RunnableService - Services with lifecycle management
• ServiceProxy - Calling services
• ServiceCluster - Managing multiple services
• Events - Publish-subscribe patterns

Advanced Topics

• Streaming RPC - Server-streaming responses
• Error Handling - Retries and DLQ
• Message Priority - Priority queues, and the migration an existing queue needs
• Custom Logger - Logging integration
• Custom Types - Extending the type system

Sample Application

The sample/combatGame directory contains a complete example - a turn-based battle royale game with 6 AI players demonstrating:

• Multiple services communicating via RPC
• Event-based game state synchronization
• Different player strategies

Run it:

python sample/combatGame/game_runner.py

Cross-Language Compatibility

This library is fully wire-compatible with the original TypeScript Protobus. Services written in Python and TypeScript can communicate seamlessly on the same message bus, thanks to Protocol Buffers' language-agnostic serialization format.

Original Project

This is the Python port of Protobus (TypeScript). The original project provides identical functionality for Node.js/TypeScript environments.

License

MIT License - Copyright (c) Remarkable Games Ltd.

See LICENSE for details.