Dramatiq

Warnings

• breaking In v2.0.0, the `ResultMiddleware` constructor now requires a `backend` argument. Previously, it would implicitly try to infer one or default.
  Fix: Pass a backend instance (e.g., `RedisBackend()`) when initializing `ResultMiddleware`: `ResultMiddleware(backend=RedisBackend())`.
• breaking In v2.0.0, the `StubBroker.join()` `fail_fast` parameter's default value changed from `False` to `True`. This means `join()` will now raise an exception immediately on task failure.
  Fix: If you relied on the previous behavior where `join()` would attempt to process all tasks regardless of individual failures, explicitly set `fail_fast=False` when calling `StubBroker.join()`.
• gotcha Using Gevent with free-threaded Python (e.g., Python 3.13+) is not recommended by Dramatiq and can lead to unexpected behavior. Dramatiq will issue a warning if this combination is detected.
  Fix: Avoid using Gevent with free-threaded Python versions. Consider alternative concurrency models or using a standard Python runtime if Gevent is critical.
• gotcha Dramatiq relies on a global broker instance configured via `dramatiq.set_broker()`. In multi-application environments, tests, or concurrent contexts, careful management is needed to ensure the correct broker is active for each operation.
  Fix: Always call `dramatiq.set_broker()` explicitly at the entry point of your application or test setup. In tests, use distinct broker instances and reset them as necessary for isolation.
• gotcha Specific broker backends (e.g., Redis, RabbitMQ) and other features (e.g., Prometheus metrics) are installed via optional 'extras' (e.g., `pip install dramatiq[redis]`). Forgetting these will result in `ModuleNotFoundError` or similar import errors.
  Fix: Ensure you install the required extras for your chosen broker and features: `pip install dramatiq[redis]` or `pip install dramatiq[rabbitmq]`.

Install

• pip install dramatiq Base installation
• pip install dramatiq[redis] With Redis broker
• pip install dramatiq[rabbitmq] With RabbitMQ broker

Imports

• actor

  from dramatiq import actor

• set_broker

  from dramatiq import set_broker

• Broker

  from dramatiq import Broker

• RedisBroker

  from dramatiq.brokers.redis import RedisBroker

• RabbitmqBroker

  from dramatiq.brokers.rabbitmq import RabbitmqBroker

• StubBroker

  from dramatiq.brokers.stub import StubBroker

• ResultMiddleware

  from dramatiq.middleware import ResultMiddleware

Quickstart

This quickstart demonstrates defining and sending a task using Dramatiq with a `StubBroker` for synchronous, in-process execution, ideal for testing. For production, you would configure a `RedisBroker` or `RabbitmqBroker` and run a separate `dramatiq worker` process to consume tasks.

import dramatiq
from dramatiq.brokers.stub import StubBroker
import time
import os

# Configure the stub broker for local testing and synchronous processing
# Note: StubBroker processes tasks directly without a separate worker process.
# For real applications, use RedisBroker, RabbitmqBroker, etc., with 'pip install dramatiq[broker]'.
broker = StubBroker()
dramatiq.set_broker(broker)

@dramatiq.actor
def my_task(name):
    print(f"[Task] Starting task for {name}...")
    time.sleep(0.01) # Simulate some work
    print(f"[Task] Task for {name} completed.")
    return f"Hello, {name}!"

# Send a message to the broker
print("Sending message...")
message = my_task.send("World")
print(f"Sent message with ID: {message.message_id}")

# With StubBroker, you can explicitly process pending messages
# In a real application, a 'dramatiq worker' process would handle this.
broker.join(drop_messages=True) # Process all pending messages
print("All stub broker messages processed.")

# If ResultMiddleware and a backend were configured, you could retrieve results:
# try:
#     result = message.get_result(block=True, timeout=1)
#     print(f"Task result: {result}")
# except Exception as e:
#     print(f"Could not get result: {e}")