### Install Dramatiq from source

Source: https://dramatiq.io/_sources/installation.rst.txt

Clones the repository and installs the package locally.

```bash
$ git clone https://github.com/Bogdanp/dramatiq
```

```bash
$ python setup.py install
```

--------------------------------

### Install Dramatiq with RabbitMQ and Requests

Source: https://dramatiq.io/_sources/guide.rst.txt

Install Dramatiq with RabbitMQ and watch support, plus the Requests library, in a new virtual environment. Requires RabbitMQ to be installed and running.

```bash
$ pip install 'dramatiq[rabbitmq, watch]' requests
```

--------------------------------

### Install Dramatiq with all features

Source: https://dramatiq.io/_sources/installation.rst.txt

Installs Dramatiq with all available extra requirements.

```bash
$ pip install -U 'dramatiq[all]'
```

--------------------------------

### Bind worker groups to queues with CLI

Source: https://dramatiq.io/cookbook.html

Command-line examples showing how to start worker groups that only consume the 'default' queue or the 'ui-blocking' queue. This allows controlling throughput by separating message processing.

```bash
# Only consume the "default" queue
$ dramatiq an_app --queues default

# Only consume the "ui-blocking" queue
$ dramatiq an_app --queues ui-blocking

```

--------------------------------

### Install Rollbar client

Source: https://dramatiq.io/_sources/cookbook.rst.txt

Install the Rollbar Python client to enable error reporting.

```bash
$ pip install rollbar
```

--------------------------------

### Install Dramatiq with RabbitMQ broker

Source: https://dramatiq.io/_sources/index.rst.txt

Installation command for Dramatiq with RabbitMQ support and the watch extra. Use this when you plan to use RabbitMQ as the message broker.

```bash
$ pip install -U 'dramatiq[rabbitmq, watch]'
```

--------------------------------

### Install Dramatiq with Redis broker

Source: https://dramatiq.io/_sources/index.rst.txt

Installation command for Dramatiq with Redis support and the watch extra. Use this when you plan to use Redis as the message broker.

```bash
$ pip install -U 'dramatiq[redis, watch]'
```

--------------------------------

### Initialize RabbitmqBroker with connection parameters

Source: https://dramatiq.io/_modules/dramatiq/brokers/rabbitmq.html

Example of specifying connection parameters individually.

```python
>>> RabbitmqBroker(host="127.0.0.1", port=5672)
```

--------------------------------

### after_consumer_thread_boot

Source: https://dramatiq.io/_modules/dramatiq/middleware/middleware.html

Called from a consumer thread after it starts but before it starts its run loop. Use this to set up resources for the consumer thread.

```APIDOC
## after_consumer_thread_boot

### Description
Called from a consumer thread after it starts but before it starts its run loop.

### Parameters
- **broker** (Broker): The broker instance.
- **thread** (ConsumerThread): The consumer thread instance.

```

--------------------------------

### Dramatiq worker log output

Source: https://dramatiq.io/_sources/guide.rst.txt

Example log output from starting a Dramatiq worker. Shows the main process booting, worker processes becoming ready, and the actor receiving and completing a message.

```text
[2017-11-19 13:03:48,188] [PID 13047] [MainThread] [dramatiq.MainProcess] [INFO] Dramatiq '0.13.1' is booting up.
[2017-11-19 13:03:48,349] [PID 22377] [MainThread] [dramatiq.WorkerProcess(3)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,350] [PID 22375] [MainThread] [dramatiq.WorkerProcess(1)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,357] [PID 22376] [MainThread] [dramatiq.WorkerProcess(2)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,357] [PID 22374] [MainThread] [dramatiq.WorkerProcess(0)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,358] [PID 22379] [MainThread] [dramatiq.WorkerProcess(5)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,362] [PID 22381] [MainThread] [dramatiq.WorkerProcess(7)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,364] [PID 22380] [MainThread] [dramatiq.WorkerProcess(6)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,366] [PID 22378] [MainThread] [dramatiq.WorkerProcess(4)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,369] [PID 22377] [Thread-4] [count_words.count_words] [INFO] Received args=('http://example.com',) kwargs={}.
There are 338 words at 'http://example.com'.
[2017-11-19 13:03:48,679] [PID 22377] [Thread-4] [count_words.count_words] [INFO] Completed after 310.42ms.
```

--------------------------------

### Initialize RabbitmqBroker with a connection URL

Source: https://dramatiq.io/_modules/dramatiq/brokers/rabbitmq.html

Example of using a connection URL to initialize the broker.

```python
>>> RabbitmqBroker(url="amqp://guest:guest@127.0.0.1:5672")
```

--------------------------------

### after_worker_thread_boot

Source: https://dramatiq.io/_modules/dramatiq/middleware/middleware.html

Called from a worker thread after it starts but before it starts its run loop. Use this to set up resources for the worker thread.

```APIDOC
## after_worker_thread_boot

### Description
Called from a worker thread after it starts but before it starts its run loop.

### Parameters
- **broker** (Broker): The broker instance.
- **thread** (WorkerThread): The worker thread instance.

```

--------------------------------

### Initialize RedisBroker with a connection URL

Source: https://dramatiq.io/reference.html

Example of initializing a RedisBroker using a standard Redis connection URL string.

```python
>>> RedisBroker(url="redis://127.0.0.1:6379/0")
```

--------------------------------

### Initialize RedisBroker with connection parameters

Source: https://dramatiq.io/reference.html

Example of initializing a RedisBroker by specifying individual connection parameters like host, port, and password.

```python
>>> RedisBroker(host="127.0.0.1", port=6379, db=0, password="hunter2")
```

--------------------------------

### Configure RabbitmqBroker with priority support

Source: https://dramatiq.io/_modules/dramatiq/brokers/rabbitmq.html

Example of setting max_priority and enqueuing messages with broker_priority.

```python
>>> broker = RabbitmqBroker(url="...", max_priority=5)
>>> broker.enqueue(an_actor.message_with_options(
...    broker_priority=5,
... ))
```

--------------------------------

### Send message with broker_priority

Source: https://dramatiq.io/_modules/dramatiq/brokers/rabbitmq.html

Example of using broker_priority with send_with_options.

```python
>>> an_actor.send_with_options(
...    broker_priority=5,
... )
```

--------------------------------

### Dramatiq Prometheus middleware initialization and setup

Source: https://dramatiq.io/_modules/dramatiq/middleware/prometheus.html

Configure DB_PATH, HTTP host and port, then define the middleware class with metric counters, gauges, and a histogram.

```python
import os
import tempfile
from http.server import BaseHTTPRequestHandler, HTTPServer

from ..common import current_millis
from ..logging import get_logger
from .middleware import Middleware

#: The path to store the prometheus database files.  This path is
#: cleared before every run.
DB_PATH = os.getenv("dramatiq_prom_db", "%s/dramatiq-prometheus" % tempfile.gettempdir())

# Ensure the DB_PATH exists.
os.makedirs(DB_PATH, exist_ok=True)

#: The HTTP host the exposition server should bind to.
HTTP_HOST = os.getenv("dramatiq_prom_host", "0.0.0.0")

#: The HTTP port the exposition server should listen on.
HTTP_PORT = int(os.getenv("dramatiq_prom_port", "9191"))




[docs]
class Prometheus(Middleware):
    """A middleware that exports stats via Prometheus_.

    .. _Prometheus: https://prometheus.io
    """

    def __init__(self) -> None:
        self.logger = get_logger(__name__, type(self))
        self.delayed_messages: set[str] = set()
        self.message_start_times: dict[str, int] = {}

```

--------------------------------

### Run Prometheus exposition server

Source: https://dramatiq.io/_modules/dramatiq/middleware/prometheus.html

Starts the HTTP server that serves the Prometheus metrics. It runs until interrupted and returns 0 on shutdown.

```python
def _run_exposition_server():
    logger = get_logger(__name__, "_run_exposition_server")
    logger.debug("Starting exposition server...")

    try:
        address = (HTTP_HOST, HTTP_PORT)
        httpd = HTTPServer(address, _metrics_handler)
        httpd.serve_forever()
    except KeyboardInterrupt:
        logger.debug("Stopping exposition server...")
        httpd.shutdown()

    return 0
```

--------------------------------

### Start Dramatiq worker for count_words

Source: https://dramatiq.io/_sources/guide.rst.txt

Starts a Dramatiq worker for the count_words module. Spawns as many processes as CPU cores with 8 worker threads per process. Use 'dramatiq -h' for available flags.

```bash
$ dramatiq count_words
```

--------------------------------

### after_worker_boot

Source: https://dramatiq.io/_modules/dramatiq/middleware/middleware.html

Called after the worker process has started up. Use this for tasks that should run once the worker is fully initialized.

```APIDOC
## after_worker_boot

### Description
Called after the worker process has started up.

### Parameters
```
broker: Broker
worker: Worker
```

```

--------------------------------

### Compose actors with message and pipeline

Source: https://dramatiq.io/reference.html

Build a message for actor composition. The example shows a pipeline of two messages.

```python
>>> (add.message(1, 2) | add.message(3))
pipeline([add(1, 2), add(3)])
```

--------------------------------

### before_worker_boot

Source: https://dramatiq.io/_modules/dramatiq/middleware/middleware.html

Called before the worker process starts up. Use this hook for initialization tasks that need to happen before the worker begins processing messages.

```APIDOC
## before_worker_boot

### Description
Called before the worker process starts up.

### Method
before_worker_boot

### Parameters
- **broker** (Broker) - The broker instance.
- **worker** (Worker) - The worker instance.

```

--------------------------------

### Provide a list of middleware classes to a broker

Source: https://dramatiq.io/_sources/advanced.rst.txt

Pass a list of middleware to the Broker constructor to replace the default middleware. This example disables Callbacks and Pipelines, sets a custom max_retries of 5 for Retries, and adds the non-default CurrentMessage middleware. Note that omitting default middleware can lead to behavior different from documented.

```python
import dramatiq
from dramatiq.middleware import AgeLimit, TimeLimit, ShutdownNotifications, Retries, CurrentMessage

# Replace `dramatiq.Broker` with a concrete class, e.g. dramatiq.brokers.redis.RedisBroker.
broker = dramatiq.Broker(
    middleware=[
        AgeLimit(),
        TimeLimit(),
        ShutdownNotifications(),
        # Note: custom default max_retries of 5
        Retries(max_retries=5),
        # Note: non-default middleware class included.
        CurrentMessage(),
        # Note: Callbacks and Pipelines are not included.
        # They will not be added back by dramatiq.
    ],
)

dramatiq.set_broker(broker)
```

--------------------------------

### Group messages and run in parallel

Source: https://dramatiq.io/cookbook.html

Create a group of messages to run actors in parallel. The example groups three messages for the frobnicate actor and runs them. Results are returned in the same order messages were added.

```python
g = group([
    frobnicate.message(1, 2),
    frobnicate.message(2, 3),
    frobnicate.message(3, 4),
]).run()

```

--------------------------------

### Prometheus context manager and metric lifecycle

Source: https://dramatiq.io/_modules/dramatiq/middleware/prometheus.html

Context manager stores delayed messages and message start times. The after_process_boot method sets the multiprocess directory, imports prometheus_client, and defines counters, gauges, and a histogram with specific metric names and labels.

```python
@property
def forks(self):
        return [_run_exposition_server]

    def after_process_boot(self, broker):
        os.environ["PROMETHEUS_MULTIPROC_DIR"] = DB_PATH
        os.environ["prometheus_multiproc_dir"] = DB_PATH

        # This import MUST happen at runtime, after process boot and
        # after the env variable has been set up.
        import prometheus_client as prom

        self.logger.debug("Setting up metrics...")
        registry = prom.CollectorRegistry()
        self.total_messages = prom.Counter(
            "dramatiq_messages_total",
            "The total number of messages processed.",
            ["queue_name", "actor_name"],
            registry=registry,
        )
        self.total_errored_messages = prom.Counter(
            "dramatiq_message_errors_total",
            "The total number of errored messages.",
            ["queue_name", "actor_name"],
            registry=registry,
        )
        self.total_retried_messages = prom.Counter(
            "dramatiq_message_retries_total",
            "The total number of retried messages.",
            ["queue_name", "actor_name"],
            registry=registry,
        )
        self.total_rejected_messages = prom.Counter(
            "dramatiq_message_rejects_total",
            "The total number of dead-lettered messages.",
            ["queue_name", "actor_name"],
            registry=registry,
        )
        self.inprogress_messages = prom.Gauge(
            "dramatiq_messages_inprogress",
            "The number of messages in progress.",
            ["queue_name", "actor_name"],
            registry=registry,
            multiprocess_mode="livesum",
        )
        self.inprogress_delayed_messages = prom.Gauge(
            "dramatiq_delayed_messages_inprogress",
            "The number of delayed messages in memory.",
            ["queue_name", "actor_name"],
            registry=registry,
        )
        self.message_durations = prom.Histogram(
            "dramatiq_message_duration_milliseconds",
            "The time spent processing messages.",
            ["queue_name", "actor_name"],
            buckets=(
                5,
                10,
                25,
                50,
                75,
                100,
                250,
                500,
                750,
                1000,
                2500,
                5000,
                7500,
                10000,
                30000,
                60000,
                600000,
                900000,
                float("inf"),
            ),
            registry=registry,
        )
```

--------------------------------

### get_broker()

Source: https://dramatiq.io/_modules/dramatiq/broker.html

Returns the global broker instance. If no global broker is set, it creates a default RabbitMQ broker, falling back to a Redis broker if RabbitMQ dependencies are not installed.

```APIDOC
## get_broker()

### Description
Returns the global broker instance. If no global broker is set, a RabbitMQ broker is created and set as the global broker. If RabbitMQ dependencies are not installed, a Redis broker is used instead.

### Method
Function

### Signature
```python
def get_broker() -> Broker
```

### Returns
- **Broker** - The default Broker instance.

### Example
```python
from dramatiq.broker import get_broker
broker = get_broker()
```
```

--------------------------------

### dramatiq.get_broker()

Source: https://dramatiq.io/reference.html

Returns the global broker instance. If no global broker is set, a RabbitMQ broker is returned if dependencies are installed, otherwise a Redis broker.

```APIDOC
## dramatiq.get_broker()

### Description
Get the global broker instance. If no global broker is set, a RabbitMQ broker will be returned. If the RabbitMQ dependencies are not installed, a Redis broker will be returned.

### Method
Function

### Endpoint
N/A

### Parameters
None

### Returns
- **Broker** - The default Broker.

### Return Type
Broker
```

--------------------------------

### Auto-discover tasks modules with bash

Source: https://dramatiq.io/_sources/cookbook.rst.txt

Use a bash script to find and register all modules named tasks.py, then start the dramatiq-gevent worker.

```bash
#!/usr/bin/env bash

set -e

tasks_packages=$(find . -type d -name tasks | sed s':/:.:g' | sed s'/^..//' | xargs)
tasks_modules=$(find . -type f -name tasks.py | sed s':/:.:g' | sed s'/^..//' | sed s'/.py$//g' | xargs)
all_modules="$tasks_packages $tasks_modules"

echo "Discovered tasks modules:"
for module in $all_modules; do
  echo "  * ${module}"
done
echo

dramatiq-gevent $all_modules --watch . --watch-use-polling
```

--------------------------------

### Worker process output

Source: https://dramatiq.io/_sources/guide.rst.txt

Example of the log output generated by the worker process when processing the tasks.

```text
[2017-11-19 13:10:02,620] [PID 24357] [Thread-4] [count_words.count_words] [INFO] Received args=('https://rabbitmq.com',) kwargs={}.
[2017-11-19 13:10:02,621] [PID 24357] [Thread-6] [count_words.count_words] [INFO] Received args=('https://xkcd.com',) kwargs={}.
[2017-11-19 13:10:02,621] [PID 24357] [Thread-5] [count_words.count_words] [INFO] Received args=('https://news.ycombinator.com',) kwargs={}.
There are 888 words at 'https://rabbitmq.com'.
[2017-11-19 13:10:02,757] [PID 24357] [Thread-4] [count_words.count_words] [INFO] Completed after 137.26ms.
There are 461 words at 'https://xkcd.com'.
[2017-11-19 13:10:02,841] [PID 24357] [Thread-6] [count_words.count_words] [INFO] Completed after 219.76ms.
There are 3598 words at 'https://news.ycombinator.com'.
[2017-11-19 13:10:03,297] [PID 24357] [Thread-5] [count_words.count_words] [INFO] Completed after 675.19ms.
```

--------------------------------

### Get the global broker instance

Source: https://dramatiq.io/_modules/dramatiq/broker.html

Retrieves the global broker instance. If none is set, it defaults to RabbitMQ if dependencies are present, otherwise falling back to Redis.

```python
def get_broker() -> Broker:
    """Get the global broker instance.

    If no global broker is set, a RabbitMQ broker will be returned.
    If the RabbitMQ dependencies are not installed, a Redis broker
    will be returned.

    Returns:
      Broker: The default Broker.
    """
    global global_broker
    if global_broker is None:
        # RabbitMQ will be tried first, but its dependencies might not
        # be installed if the user only installed the [redis] extras.
        try:
            from .brokers.rabbitmq import RabbitmqBroker

            set_broker(
                RabbitmqBroker(
                    host="127.0.0.1",
                    port=5672,
                    heartbeat=5,
                    connection_attempts=5,
                    blocked_connection_timeout=30,
                )
            )
        except ImportError:
            # Fall back to the Redis broker.
            from .brokers.redis import RedisBroker

            set_broker(RedisBroker())
    global_broker = cast(Broker, global_broker)
    return global_broker
```

--------------------------------

### Define actors for pipeline example

Source: https://dramatiq.io/cookbook.html

Define two actors: one that fetches URI contents and another that counts words in the text. The second actor receives the URI and the text from the first actor.

```python
@dramatiq.actor
def get_uri_contents(uri):
    return requests.get(uri).text


@dramatiq.actor
def count_words(uri, text):
    count = len(text.split(" "))
    print(f"There are {count} words at {uri}.")

```

--------------------------------

### Add extra middleware to a broker

Source: https://dramatiq.io/_sources/advanced.rst.txt

Use broker.add_middleware() to add extra middleware while keeping the default middleware unchanged. This example adds the CurrentMessage middleware.

```python
from dramatiq import Broker
from dramatiq.middleware import CurrentMessage

broker = Broker()
broker.add_middleware(CurrentMessage())
```

--------------------------------

### count_words.py - Basic URL word counter

Source: https://dramatiq.io/_sources/guide.rst.txt

A simple function that fetches a URL and prints the number of space-separated chunks in the response. Used as the starting point for the Dramatiq actor example.

```python
import requests


def count_words(url):
    response = requests.get(url)
    count = len(response.text.split(" "))
    print(f"There are {count} words at {url!r}.")
```

--------------------------------

### Initialize MemcachedBackend

Source: https://dramatiq.io/_modules/dramatiq/rate_limits/backends/memcached.html

Demonstrates how to import and instantiate the MemcachedBackend with server configuration.

```python
>>> from dramatiq.rate_limits.backends import MemcachedBackend
>>> backend = MemcachedBackend(servers=["127.0.0.1"], binary=True)
```

--------------------------------

### Set actor priority with the priority keyword argument

Source: https://dramatiq.io/guide.html

Define actors with different priority values. Lower numeric values indicate higher priority. The default priority is 0. This example shows two actors with priorities 0 and 10.

```python
@dramatiq.actor(priority=0)  # 0 is the default
def generate_report(user_id):
    ...


@dramatiq.actor(priority=10)
def sync_order_to_warehouse(order_id):
    ...
```

--------------------------------

### Unit Testing with StubBroker

Source: https://dramatiq.io/_sources/guide.rst.txt

Use StubBroker in unit tests to avoid requiring a running RabbitMQ or Redis instance. The example shows configuring the broker in broker.py and defining pytest fixtures in conftest.py.

```python
import os

from dramatiq.brokers.rabbitmq import RabbitmqBroker
from dramatiq.brokers.stub import StubBroker


if os.getenv("UNIT_TESTS") == "1":
    broker = StubBroker()
    broker.emit_after("process_boot")
else:
    broker = RabbitmqBroker()
```

```python
import dramatiq
import pytest

from dramatiq import Worker
from yourapp import broker


@pytest.fixture()
def stub_broker():
    broker.flush_all()
    return broker


@pytest.fixture()
def stub_worker():
    worker = Worker(broker, worker_timeout=100)
    worker.start()
    yield worker
    worker.stop()
```

--------------------------------

### Worker error log for failed message

Source: https://dramatiq.io/guide.html

Example log output when an actor raises an unhandled exception. Shows the warning, traceback, and retry scheduling with delay.

```text
[2017-06-27 13:11:22,059] [PID 13053] [Thread-8] [dramatiq.worker.WorkerThread] [WARNING] Failed to process message count_words('foo') with unhandled exception.
Traceback (most recent call last):
  ...
requests.exceptions.MissingSchema: Invalid URL 'foo': No schema supplied. Perhaps you meant http://foo?
[2017-06-27 13:11:22,062] [PID 13053] [Thread-8] [dramatiq.middleware.retries.Retries] [INFO] Retrying message 'a53a5a7d-74e1-48ae-a5a8-0b72af2a8708' in 8104 milliseconds.

```

--------------------------------

### Define DynamicTimeLimitMiddleware with before_process_message hook

Source: https://dramatiq.io/cookbook.html

Subclass dramatiq.Middleware and override before_process_message to set the message time_limit option dynamically based on message arguments. The calculate_time_limit function is referenced but not defined in the example.

```python
import dramatiq

class DynamicTimeLimitMiddleware(dramatiq.Middleware):
  def before_process_message(self, broker, message):
      """Sets a dynamic time limit on messages based on their arguments."""
      message.options["time_limit"] = calculate_time_limit(
          message.args, message.kwargs
      )

```

--------------------------------

### RateLimiter class source code

Source: https://dramatiq.io/_modules/dramatiq/rate_limits/rate_limiter.html

The complete source code for the RateLimiter class, including the license header, imports, class definition, docstring with usage examples, and the acquire context manager.

```python
# This file is a part of Dramatiq.
#
# Copyright (C) 2017,2018 CLEARTYPE SRL <bogdan@cleartype.io>
#
# Dramatiq is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# Dramatiq is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from __future__ import annotations

from contextlib import contextmanager

from ..errors import RateLimitExceeded




[docs]
class RateLimiter:
    """ABC for rate limiters.

    Examples:

      >>> from dramatiq.rate_limits.backends import RedisBackend

      >>> backend = RedisBackend()
      >>> limiter = ConcurrentRateLimiter(backend, "distributed-mutex", limit=1)

      >>> with limiter.acquire(raise_on_failure=False) as acquired:
      ...     if not acquired:
      ...         print("Mutex not acquired.")
      ...         return
      ...
      ...     print("Mutex acquired.")

    Parameters:
      backend(RateLimiterBackend): The rate limiting backend to use.
      key(str): The key to rate limit on.
    """

    def __init__(self, backend, key):
        self.backend = backend
        self.key = key

    def _acquire(self):  # pragma: no cover
        raise NotImplementedError

    def _release(self):  # pragma: no cover
        raise NotImplementedError



[docs]
    @contextmanager
    def acquire(self, *, raise_on_failure=True):
        """Attempt to acquire a slot under this rate limiter.

        Parameters:
          raise_on_failure(bool): Whether or not failures should raise an
            exception.  If this is false, the context manager will instead
            return a boolean value representing whether or not the rate
            limit slot was acquired.

        Returns:
          bool: Whether or not the slot could be acquired.
        """
        acquired = False

        try:
            acquired = self._acquire()
            if raise_on_failure and not acquired:
                raise RateLimitExceeded("rate limit exceeded for key %(key)r" % vars(self))

            yield acquired
        finally:
            if acquired:
                self._release()

```

--------------------------------

### Get pipeline end result

Source: https://dramatiq.io/cookbook.html

Get the result of the last actor in a pipeline, blocking until it is available with a timeout.

```python
pipe.get_result(block=True, timeout=5_000)

```

--------------------------------

### RedisBackend.__init__

Source: https://dramatiq.io/_modules/dramatiq/results/backends/redis.html

Initializes a RedisBackend instance with configurable namespace, encoder, key prefixing, client, and connection parameters.

```APIDOC
## RedisBackend.__init__

### Description
Initializes a RedisBackend instance. This constructor sets up the backend with a namespace, encoder, and optional client or URL. If a client is provided, all other parameters are ignored. If a URL is provided, it is used to create a connection pool.

### Method
Constructor

### Signature
RedisBackend(*, namespace="dramatiq-results", encoder=None, use_namespace_prefix_keys=False, client=None, url=None, **parameters)

### Parameters
- **namespace** (str) - Optional - A string with which to prefix result keys. Defaults to "dramatiq-results".
- **encoder** (Encoder) - Optional - The encoder to use when storing and retrieving result data. Defaults to JSONEncoder.
- **use_namespace_prefix_keys** (bool) - Optional - When True, keys use the format <namespace>:<queue>:<actor>:<message_id> for human-readable keys. When False (default), keys are MD5 hashes of that format.
- **client** (Redis) - Optional - An optional Redis client. If passed, all other parameters are ignored.
- **url** (str) - Optional - An optional connection URL. If both URL and connection parameters are provided, the URL is used.
- **parameters** - Optional - Connection parameters passed directly to redis.Redis.

### Returns
None
```

--------------------------------

### get_declared_queues

Source: https://dramatiq.io/_modules/dramatiq/broker.html

Gets all declared queues.

```APIDOC
## get_declared_queues

### Description
Gets all declared queues.

### Method
get_declared_queues

### Parameters
None

### Returns
- **set[str]** - The names of all the queues declared so far on this Broker.
```

--------------------------------

### get_declared_actors

Source: https://dramatiq.io/_modules/dramatiq/broker.html

Gets all declared actors.

```APIDOC
## get_declared_actors

### Description
Gets all declared actors.

### Method
get_declared_actors

### Parameters
None

### Returns
- **set[str]** - The names of all the actors declared so far on this Broker.
```

--------------------------------

### get_declared_delay_queues()

Source: https://dramatiq.io/reference.html

Gets all declared delay queues.

```APIDOC
## get_declared_delay_queues()

### Description
Gets all declared delay queues.

### Method
Function

### Parameters
None

### Returns
set[str] - The names of all the delay queues declared so far on this Broker.
```

--------------------------------

### Configure Results Middleware

Source: https://dramatiq.io/reference.html

Initialize the Redis backend and add the Results middleware to the broker.

```python
>>> from dramatiq.results import Results
>>> from dramatiq.results.backends import RedisBackend
>>> backend = RedisBackend()
>>> broker.add_middleware(Results(backend=backend))
```

--------------------------------

### RedisBroker.__init__

Source: https://dramatiq.io/_modules/dramatiq/brokers/redis.html

Initializes a RedisBroker instance with optional connection parameters, middleware, and configuration options.

```APIDOC
## RedisBroker.__init__

### Description
Initializes a RedisBroker instance. Accepts either a connection URL or individual connection parameters, along with middleware and configuration options.

### Method
Constructor

### Parameters
#### Keyword Arguments
- **url** (str) - Optional - An optional connection URL. If both a URL and connection parameters are provided, the URL is used.
- **middleware** (list[Middleware]) - Optional - A list of middleware to apply.
- **namespace** (str) - Optional - The string with which to prefix all Redis keys. Defaults to "dramatiq".
- **maintenance_chance** (int) - Optional - How many commands out of a million should trigger queue maintenance. Defaults to 1000.
- **heartbeat_timeout** (int) - Optional - The amount of time (in ms) that has to pass without a heartbeat for a worker to be considered offline. Defaults to 60000.
- **dead_message_ttl** (int) - Optional - The amount of time (in ms) that dead-lettered messages are kept in Redis for. Defaults to 86400000 * 7.
- **client** (redis.Redis) - Optional - A redis client to use.
- **parameters** - Optional - Connection parameters are passed directly to redis.Redis.

### Example
RedisBroker(host="127.0.0.1", port=6379, db=0, password="hunter2")

RedisBroker(url="redis://127.0.0.1:6379/0")
```

--------------------------------

### Broker.get_results_backend

Source: https://dramatiq.io/reference.html

Gets the backend of the Results middleware.

```APIDOC
## Broker.get_results_backend

### Description
Get the backend of the Results middleware.

### Method
get_results_backend

### Parameters
None

### Raises
- **RuntimeError** - If the broker doesn't have a results backend.

### Returns
The backend (ResultBackend).
```

--------------------------------

### Set up Prometheus multiproc environment for Dramatiq

Source: https://dramatiq.io/advanced.html

Creates a temporary directory, clears it, and sets prometheus_multiproc_dir and dramatiq_prom_db environment variables before running dramatiq app. Required to avoid Prometheus multi-process export issues.

```bash
mkdir -p /tmp/dramatiq-prometheus \
  && rm -r /tmp/dramatiq-prometheus/* \
  && env prometheus_multiproc_dir=/tmp/dramatiq-prometheus \
         dramatiq_prom_db=/tmp/dramatiq-prometheus \
         dramatiq app

```

--------------------------------

### Pipeline.get_results

Source: https://dramatiq.io/reference.html

Gets the results of each job in the pipeline.

```APIDOC
## Pipeline.get_results

### Description
Gets the results of each job in the pipeline.

### Method
Pipeline.get_results(*, block=False, timeout=None)

### Parameters
- **block** (bool) - Optional - Whether or not to block until a result is set.
- **timeout** (int) - Optional - The maximum amount of time, in ms, to wait for a result when block is True. Defaults to 10 seconds.

### Returns
- **generator** - A result generator.

### Raises
- **ResultMissing** - When block is False and the result isn't set.
- **ResultTimeout** - When waiting for a result times out.
```

--------------------------------

### Broker.dead_letters

Source: https://dramatiq.io/reference.html

Gets the dead-lettered messages for all defined queues.

```APIDOC
## dead_letters (property)

### Description
The dead-lettered messages for all defined queues.
```

--------------------------------

### consume

Source: https://dramatiq.io/_modules/dramatiq/broker.html

Gets an iterator that consumes messages off of the queue.

```APIDOC
## consume

### Description
Gets an iterator that consumes messages off of the queue.

### Method
consume

### Parameters
- **queue_name** (str) - Required - The name of the queue to consume messages off of.
- **prefetch** (int) - Optional - The number of messages to prefetch per consumer. Default: 1.
- **timeout** (int) - Optional - The amount of time in milliseconds to idle for. Default: 30000.

### Returns
- **Consumer** - A message iterator.

### Raises
- **QueueNotFound** - If the given queue was never declared.
```

--------------------------------

### StubBroker.__init__

Source: https://dramatiq.io/_modules/dramatiq/brokers/stub.html

Initializes a new StubBroker instance with optional middleware and a fail_fast_default flag.

```APIDOC
## StubBroker.__init__

### Description
Initializes a new StubBroker instance.

### Method
Constructor

### Parameters
- **middleware** (list[Middleware], optional) - Optional - List of middleware to use.
- **fail_fast_default** (bool, optional) - Optional - Default value for the fail_fast argument of the join method. Defaults to True.

### Returns
StubBroker - A new StubBroker instance.
```

--------------------------------

### Broker.consume

Source: https://dramatiq.io/reference.html

Gets an iterator that consumes messages off of the specified queue.

```APIDOC
## Broker.consume

### Description
Get an iterator that consumes messages off of the queue.

### Method
consume

### Parameters
- **queue_name** (str) - Required - The name of the queue to consume messages off of.
- **prefetch** (int) - Optional - The number of messages to prefetch per consumer. Default: 1.
- **timeout** (int) - Optional - The amount of time in milliseconds to idle for. Default: 30000.

### Raises
- **QueueNotFound** - If the given queue was never declared.

### Returns
A message iterator (Consumer).
```

--------------------------------

### Configure RabbitMQ Broker

Source: https://dramatiq.io/_sources/guide.rst.txt

Instantiate a RabbitmqBroker and set it as the global broker early in the program execution.

```python
import dramatiq

from dramatiq.brokers.rabbitmq import RabbitmqBroker


rabbitmq_broker = RabbitmqBroker(host="rabbitmq")
dramatiq.set_broker(rabbitmq_broker)
```

--------------------------------

### Add worker thread

Source: https://dramatiq.io/_modules/dramatiq/worker.html

Creates and starts a WorkerThread, then appends it to the worker list.

```python
worker = WorkerThread(
    broker=self.broker,
    consumers=self.consumers,
    work_queue=self.work_queue,
    worker_timeout=self.worker_timeout,
)
worker.start()
self.workers.append(worker)
```

--------------------------------

### Configure Redis Broker

Source: https://dramatiq.io/_sources/guide.rst.txt

Instantiate a RedisBroker and set it as the global broker early in the program execution.

```python
import dramatiq

from dramatiq.brokers.redis import RedisBroker


redis_broker = RedisBroker(host="redis")
dramatiq.set_broker(redis_broker)
```

--------------------------------

### RabbitmqBroker(confirm_delivery=False, url=None, middleware=None, max_priority=None, parameters=None, **kwargs)

Source: https://dramatiq.io/reference.html

Creates a RabbitMQ broker instance. Supports connection via URL or individual parameters, and optional queue priority support.

```APIDOC
## RabbitmqBroker(confirm_delivery=False, url=None, middleware=None, max_priority=None, parameters=None, **kwargs)

### Description
A broker that can be used with RabbitMQ. Supports specifying connection parameters individually or via a connection URL. Optionally supports queued message priorities.

### Method
Constructor

### Parameters
- **confirm_delivery** (_bool_) - Optional - Wait for RabbitMQ to confirm that messages have been committed on every call to enqueue. Must be enabled for Dramatiq to detect and re-declare missing queues when enqueuing messages. Defaults to False.
- **url** (_str | list[str]_) - Optional - An optional connection URL. If both a URL and connection parameters are provided, the URL is used.
- **middleware** (_list[Middleware]_) - Optional - The set of middleware that apply to this broker.
- **max_priority** (_int_) - Optional - Configure queues with x-max-priority to support queue-global priority queueing.
- **parameters** (_list[dict]_) - Optional - A sequence of (pika) connection parameters to determine which Rabbit server(s) to connect to.
- **kwargs** (_Any_) - Optional - The (pika) connection parameters to use to determine which Rabbit server to connect to.

### Examples
```
>>> RabbitmqBroker(host="127.0.0.1", port=5672)
```
```
>>> RabbitmqBroker(url="amqp://guest:guest@127.0.0.1:5672")
```
```
>>> broker = RabbitmqBroker(url="...", max_priority=5)
```

```

--------------------------------

### RabbitmqBroker.connection

Source: https://dramatiq.io/reference.html

Gets the pika.BlockingConnection for the current thread. This property may change without notice.

```APIDOC
## RabbitmqBroker.connection

### Description
The pika.BlockingConnection for the current thread. This property may change without notice.

### Property
connection

```

--------------------------------

### Configure Prometheus for multi-process

Source: https://dramatiq.io/_sources/advanced.rst.txt

Prepares a directory for Prometheus metrics and runs the Dramatiq application with the necessary environment variables to avoid export failures.

```bash
mkdir -p /tmp/dramatiq-prometheus \
  && rm -r /tmp/dramatiq-prometheus/* \
  && env prometheus_multiproc_dir=/tmp/dramatiq-prometheus \
         dramatiq_prom_db=/tmp/dramatiq-prometheus \
         dramatiq app
```

--------------------------------

### RabbitmqBroker.channel

Source: https://dramatiq.io/reference.html

Gets the pika.BlockingChannel for the current thread. This property may change without notice.

```APIDOC
## RabbitmqBroker.channel

### Description
The pika.BlockingChannel for the current thread. This property may change without notice.

### Property
channel

```

--------------------------------

### Initialize BucketRateLimiter for 1 operation per minute

Source: https://dramatiq.io/_modules/dramatiq/rate_limits/bucket.html

Creates a rate limiter allowing 1 operation every 60,000 milliseconds.

```python
>>> BucketRateLimiter(backend, "some-key", limit=1, bucket=60_000)
```

--------------------------------

### Get declared queues

Source: https://dramatiq.io/_modules/dramatiq/brokers/rabbitmq.html

Returns the set of all queue names currently declared on the broker.

```python
    def get_declared_queues(self) -> set[str]:
        """Get all declared queues.

        Returns:
          set[str]: The names of all the queues declared so far on
          this Broker.
        """
        return self.queues.copy()
```

--------------------------------

### Initialize Rollbar middleware

Source: https://dramatiq.io/_sources/cookbook.rst.txt

Initialize the Rollbar client and add the custom middleware to the broker.

```python
rollbar.init(YOUR_ROLLBAR_KEY)
broker.add_middleware(path.to.RollbarMiddleware())
```

--------------------------------

### pipeline.get_result

Source: https://dramatiq.io/_modules/dramatiq/composition.html

Gets the result of the pipeline, which is the result of the last message in the chain. Can block until a result is available.

```APIDOC
## pipeline.get_result

### Description
Gets the result of the pipeline, which is the result of the last message in the chain. Can block until a result is available.

### Method
get_result

### Parameters
- **block** (bool) - Optional - Whether or not to block until a result is set.
- **timeout** (int) - Optional - The maximum amount of time, in ms, to wait for a result when block is True. Defaults to 10 seconds.

### Returns
- **object** - The result.

### Raises
- **ResultMissing** - When block is False and the result isn't set.
- **ResultTimeout** - When waiting for a result times out.
```

--------------------------------

### Create RabbitMQ virtual hosts

Source: https://dramatiq.io/_sources/advanced.rst.txt

Use rabbitmqctl to create a virtual host and set permissions for a user. This is needed for multi-tenancy with RabbitMQ.

```shell
$ rabbitmqctl add_vhost app1
$ rabbitmqctl set_permissions -p app1 my_user ".*" ".*" ".*"
```

--------------------------------

### get_queue_message_counts(queue_name)

Source: https://dramatiq.io/reference.html

Gets the number of messages in a queue. This method is only meant to be used in unit and integration tests.

```APIDOC
## get_queue_message_counts(queue_name)

### Description
Gets the number of messages in a queue. This method is only meant to be used in unit and integration tests.

### Method
Function

### Parameters
- **queue_name** (str) - Required - The queue whose message counts to get.

### Returns
tuple - A triple representing the number of messages in the queue, its delayed queue and its dead letter queue.
```

--------------------------------

### Create RabbitMQ virtual host with rabbitmqctl

Source: https://dramatiq.io/advanced.html

Use rabbitmqctl to add a virtual host and set permissions for a user. The set_permissions command grants configure, write, and read permissions (.*) on the app1 vhost.

```shell
$ rabbitmqctl add_vhost app1
$ rabbitmqctl set_permissions -p app1 my_user ".*" ".*" ".*"

```

--------------------------------

### Get pipeline intermediate results

Source: https://dramatiq.io/cookbook.html

Iterate over the intermediate results of each step in a pipeline, blocking until results are available.

```python
for res in pipe.get_results(block=True):
    ...

```

--------------------------------

### Pipeline.run

Source: https://dramatiq.io/reference.html

Runs this pipeline.

```APIDOC
## Pipeline.run

### Description
Runs this pipeline.

### Method
Pipeline.run(*, delay=None)

### Parameters
- **delay** (int) - Optional - The minimum amount of time, in milliseconds, the pipeline should be delayed by. If both pipeline's delay and first message's delay are provided, the bigger value will be used.

### Returns
- **pipeline** - Itself.
```

--------------------------------

### Configure broker.py for Unit Testing

Source: https://dramatiq.io/guide.html

Use StubBroker when the UNIT_TESTS environment variable is set to 1, otherwise default to RabbitmqBroker.

```python
import os

from dramatiq.brokers.rabbitmq import RabbitmqBroker
from dramatiq.brokers.stub import StubBroker


if os.getenv("UNIT_TESTS") == "1":
    broker = StubBroker()
    broker.emit_after("process_boot")
else:
    broker = RabbitmqBroker()
```

--------------------------------

### pipeline.__init__

Source: https://dramatiq.io/_modules/dramatiq/composition.html

Initializes a pipeline with a sequence of messages or child pipelines. Child pipelines are flattened into the resulting pipeline.

```APIDOC
## pipeline.__init__

### Description
Initializes a pipeline with a sequence of messages or child pipelines. Child pipelines are flattened into the resulting pipeline.

### Method
Constructor

### Parameters
- **children** (Iterable[Message|pipeline]) - Required - A sequence of messages or pipelines. Child pipelines are flattened into the resulting pipeline.
- **broker** (Broker) - Optional - The broker to run the pipeline on. Defaults to the current global broker.

### Returns
None
```

--------------------------------

### Initialize BucketRateLimiter for 10 operations per second

Source: https://dramatiq.io/_modules/dramatiq/rate_limits/bucket.html

Creates a rate limiter allowing up to 10 operations every 1,000 milliseconds.

```python
>>> BucketRateLimiter(backend, "some-key", limit=10, bucket=1_000)
```

--------------------------------

### Declare an actor with @dramatiq.actor

Source: https://dramatiq.io/reference.html

Decorator to declare an actor. The example shows the resulting Actor object with default queue and actor name.

```python
>>> @dramatiq.actor
... def add(x, y):
...     print(x + y)
...
>>> add
Actor(<function add at 0x106c6d488>, queue_name='default', actor_name='add')
```

--------------------------------

### ResultBackend class definition and initialization

Source: https://dramatiq.io/_modules/dramatiq/results/backend.html

Defines the ResultBackend base class with parameters for namespace, encoder, and use_namespace_prefix_keys. The constructor imports get_encoder and sets instance attributes.

```python
class ResultBackend:
    """ABC for result backends.

    Parameters:
      namespace(str): The logical namespace under which the data
        should be stored.
      encoder(Encoder): The encoder to use when storing and retrieving
        result data.  Defaults to :class:`.JSONEncoder`.
      use_namespace_prefix_keys(bool): When True, the keys used to store results
        use the format ``<namespace>:<queue>:<actor>:<message_id>`` so that the
        keys are human-readable and can be scanned or expired by namespace in the backend.
        When False (the default) the keys are MD5 hashes of the format described above,
        so the namespace is not visible in the stored keys.
    """

    def __init__(
        self,
        *,
        namespace: str = "dramatiq-results",
        encoder: typing.Optional[Encoder] = None,
        use_namespace_prefix_keys: bool = False,
    ):
        from ..message import get_encoder

        self.namespace = namespace
        self.encoder = encoder or get_encoder()
        self.use_namespace_prefix_keys = use_namespace_prefix_keys
```

--------------------------------

### pipeline.run

Source: https://dramatiq.io/_modules/dramatiq/composition.html

Runs the pipeline by enqueuing the first message to the broker. Optionally delays the pipeline execution.

```APIDOC
## pipeline.run

### Description
Runs the pipeline by enqueuing the first message to the broker. Optionally delays the pipeline execution.

### Method
run

### Parameters
- **delay** (int) - Optional - The minimum amount of time, in milliseconds, the pipeline should be delayed by. If both pipeline's delay and first message's delay are provided, the bigger value will be used.

### Returns
- **pipeline** - Itself.
```

--------------------------------

### Get queue message counts

Source: https://dramatiq.io/_modules/dramatiq/brokers/rabbitmq.html

Retrieves the count of messages in the main, delayed, and dead letter queues. Intended for use in unit and integration tests.

```python
    def get_queue_message_counts(self, queue_name: str) -> tuple[int, int, int]:
        """Get the number of messages in a queue.  This method is only
        meant to be used in unit and integration tests.

        Parameters:
          queue_name(str): The queue whose message counts to get.

        Returns:
          tuple: A triple representing the number of messages in the
          queue, its delayed queue and its dead letter queue.
        """
        queue_response = self._declare_queue(queue_name)
        dq_queue_response = self._declare_dq_queue(queue_name)
        xq_queue_response = self._declare_xq_queue(queue_name)
        return (
            queue_response.method.message_count,
            dq_queue_response.method.message_count,
            xq_queue_response.method.message_count,
        )
```

--------------------------------

### Register custom middleware

Source: https://dramatiq.io/_sources/cookbook.rst.txt

Instantiate the custom middleware and add it to the broker configuration.

```python
import dramatiq.middleware
```

--------------------------------

### Pipeline with pipe_ignore option

Source: https://dramatiq.io/cookbook.html

Use pipe_ignore=True to prevent the result of the previous actor from being passed to the next one. In this example, the result of bust_caches is not passed to prepare_codes, but the result of prepare_codes is passed to launch_missiles.

```python
(
    bust_caches.message() |
    prepare_codes.message_with_options(pipe_ignore=True) |
    launch_missiles.message()
)

```