Dramatiq

https://dramatiq.io
Admin
Dramatiq is a background task processing library for Python with a focus on simplicity, reliability,...
Tokens:110,018
Snippets:863
Trust Score:8.7
Update:1 month ago
Show doc for...
Tokens:
Raw
### 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.

```

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

### 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")
```

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

### 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.
```

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

### 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
```

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

### 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.

```

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

### 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)
```

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

### 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.
```

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

### 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
```

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

### 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))
```

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

### 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.
```

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

### 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)
```

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

### 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)
```

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

### 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)
```

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

### 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)
```
Tokens:
Raw