dramatiq-eager-broker 0.3.0

Eager broker for dramatiq

dramatiq-eager-broker

An eager broker for Dramatiq that executes tasks synchronously and immediately, without queuing. Perfect for testing and development environments.

Features

• Synchronous task execution
• No message broker required (Redis, RabbitMQ, etc.)
• Pipeline support
• Middleware support
• Drop-in replacement for testing

Installation

pip install dramatiq-eager-broker

Or with uv:

uv add dramatiq-eager-broker

Usage

import dramatiq
from dramatiq_eager_broker import EagerBroker

broker = EagerBroker(middleware=[])
dramatiq.set_broker(broker)

@dramatiq.actor
def send_email(email, message):
    print(f"Sending email to {email}: {message}")

# Tasks are executed immediately and synchronously
send_email.send("[e-mail]", "Hello!")

Testing Example

import dramatiq
import pytest
from dramatiq_eager_broker import EagerBroker


@pytest.fixture
def eager_broker():
    broker = EagerBroker(middleware=[])
    dramatiq.set_broker(broker)
    yield broker
    dramatiq.set_broker(None)


def test_my_actor(eager_broker):
    results = []

    @dramatiq.actor
    def my_task(value):
        results.append(value)

    my_task.send("test")
    assert results == ["test"]

Pipeline Support

The eager broker supports Dramatiq pipelines:

from dramatiq.middleware import Pipelines

@pytest.fixture
def eager_broker():
    broker = EagerBroker(middleware=[Pipelines()])
    dramatiq.set_broker(broker)
    yield broker
    dramatiq.set_broker(None)

@dramatiq.actor
def add(x, y):
    return x + y

@dramatiq.actor
def multiply(result, factor):
    return result * factor

# Create and execute a pipeline
pipeline = add.message(2, 3) | multiply.message(factor=10)
broker.enqueue(pipeline.messages[0])

Results Support

The eager broker supports Dramatiq get_result() if the Results middleware is loaded:

from dramatiq.middleware import Pipelines
from dramatiq.results import Results
from dramatiq.results.backends import StubBackend

@pytest.fixture
def eager_broker():
    broker = EagerBroker(middleware=[Pipelines(), Results(backend=StubBackend())])
    dramatiq.set_broker(broker)
    yield broker
    dramatiq.set_broker(None)

@dramatiq.actor(store_result=True)
def add(x, y):
    return x + y

@dramatiq.actor
def multiply(result, factor):
    return result * factor


# Create and execute a pipeline
pipeline = add.message(2, 3) | multiply.message(factor=10)
broker.enqueue(pipeline.messages[0])

assert pipeline.get_result() == 60

Results with exceptions

If the task should be executed synchronously but the actor should not raise an exception, use fail_fast=False. This mimics the conventional Dramatiq behavior which stuffs the exception into the message.

from dramatiq.middleware import Pipelines, Retries
from dramatiq.results import Results, ResultFailure
from dramatiq.results.backends import StubBackend

@pytest.fixture
def eager_broker():
    broker = EagerBroker(middleware=[Pipelines(), Retries(), Results(backend=StubBackend())], fail_fast=False)
    dramatiq.set_broker(broker)
    yield broker
    dramatiq.set_broker(None)

@dramatiq.actor(store_result=True, throws=ValueError)
def add(x, y):
    if x < y:
        raise ValueError("x < y")

message = add.send(2, 3)
with pytest.raises(ResultFailure):
    message.get_result()