Remoulade

https://remoulade.readthedocs.io/en/latest/
Admin
Remoulade is a distributed task processing library for Python with a focus on simplicity,...
Tokens:108,214
Snippets:1,138
Trust Score:9.6
Update:3 months ago
Show doc for...
Tokens:
Raw
### Install Remoulade from Source

Source: https://remoulade.readthedocs.io/en/latest/_sources/installation.rst.txt

After cloning the repository, run this command within the cloned directory to install Remoulade into your local site-packages. This requires Python setup tools.

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

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

### Configure and Start Weather Data Scheduler

Source: https://remoulade.readthedocs.io/en/latest/_sources/getting_started.rst.txt

Set up a scheduler to periodically run an actor for updating weather data. This example updates weather every 10 seconds.

```python
import requests
import remoulade
from remoulade.brokers.rabbitmq import RabbitmqBroker
from remoulade.scheduler import ScheduledJob, Scheduler

url_endpoint = "https://www.<unique-id>.endpoints.dev/"  # put your unique endpoint here

@remoulade.actor
def get_weather(city):
     url = f"https://goweather.herokuapp.com/weather/{city}"

     response = requests.get(url).json()

     text = f'{city}: {response["description"]}'
     requests.post(url_endpoint, json=text)


rabbitmq_broker = RabbitmqBroker()
remoulade.set_broker(rabbitmq_broker)
scheduler = Scheduler(rabbitmq_broker, [ScheduledJob(actor_name="get_weather", args=("Paris",), interval=10)])
remoulade.set_scheduler(scheduler)
remoulade.declare_actors([get_weather])
scheduler.start()
```

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

### Install Remoulade with All Features

Source: https://remoulade.readthedocs.io/en/latest/_sources/installation.rst.txt

Installs Remoulade with all available extra requirements. This is a comprehensive installation for all features.

```bash
pip install -U 'remoulade[all]'
```

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

### Remoulade Worker Boot-up Logs

Source: https://remoulade.readthedocs.io/en/latest/guide.html

Example log output observed when Remoulade workers start and become ready for tasks. This includes information about the Remoulade version, process IDs, and worker readiness.

```log
[2017-11-19 13:03:48,188] [PID 22370] [MainThread] [remoulade.MainProcess] [INFO] Remoulade '0.13.1' is booting up.
[2017-11-19 13:03:48,349] [PID 22377] [MainThread] [remoulade.WorkerProcess(3)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,350] [PID 22375] [MainThread] [remoulade.WorkerProcess(1)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,357] [PID 22376] [MainThread] [remoulade.WorkerProcess(2)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,357] [PID 22374] [MainThread] [remoulade.WorkerProcess(0)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,358] [PID 22379] [MainThread] [remoulade.WorkerProcess(5)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,362] [PID 22381] [MainThread] [remoulade.WorkerProcess(7)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,364] [PID 22380] [MainThread] [remoulade.WorkerProcess(6)] [INFO] Worker process is ready for action.
[2017-11-19 13:03:48,366] [PID 22378] [MainThread] [remoulade.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.

```

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

### Install Remoulade and Requests

Source: https://remoulade.readthedocs.io/en/latest/_sources/getting_started.rst.txt

Install the Remoulade library with RabbitMQ support and the Requests library for making HTTP calls.

```bash
pip install -U 'remoulade[rabbitmq]' requests
```

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

### Install Remoulade from Source

Source: https://remoulade.readthedocs.io/en/latest/installation.html

After cloning the repository, run this command within the cloned directory to install Remoulade locally.

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

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

### Install Remoulade with RabbitMQ and Requests

Source: https://remoulade.readthedocs.io/en/latest/guide.html

Install Remoulade with RabbitMQ support and the Requests library using pip.

```bash
$ pip install 'remoulade[rabbitmq]' requests

```

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

### Worker Start Sequence

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/worker.html

Starts the worker by emitting a 'worker_boot' event, adding middleware, and initializing worker threads.

```python
def start(self):
    """Initialize the worker boot sequence and start up all the
    worker threads.
    """
    self.broker.emit_before("worker_boot", self)

    worker_middleware = _WorkerMiddleware(self)
    self.broker.add_middleware(worker_middleware)
    for _ in range(self.worker_threads):
        self._add_worker()

    self.broker.emit_after("worker_boot", self)

```

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

### Setup Broker and Results Backend

Source: https://remoulade.readthedocs.io/en/latest/_sources/cookbook.rst.txt

Configure RabbitMQ as the broker and Redis as the backend for storing and retrieving task results. This setup is necessary for any task that needs to return a value or be tracked.

```python
from remoulade.brokers.rabbitmq import RabbitmqBroker
from remoulade.results.backends import RedisBackend
from remoulade.results import Results

result_backend = RedisBackend()
broker = RabbitmqBroker()
broker.add_middleware(Results(backend=result_backend))
remoulade.set_broker(broker)

@remoulade.actor(store_results=True)
def add(x, y):
  return x + y

broker.declare_actor(add)

if __name__ == "__main__":
  message = add.send(1, 2)
  print(message.result.get(block=True, raise_on_error=True, forget=False))
```

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

### Install Remoulade with Redis

Source: https://remoulade.readthedocs.io/en/latest/getting_started.html

Install the necessary dependencies for Remoulade with Redis support using pip.

```bash
$ pip install -U 'remoulade[redis]'
```

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

### Install Remoulade with RabbitMQ Support

Source: https://remoulade.readthedocs.io/en/latest/_sources/installation.rst.txt

Use this command to install Remoulade with RabbitMQ dependencies. Ensure you have pip installed.

```bash
pip install -U 'remoulade[rabbitmq]'
```

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

### Install Remoulade with RabbitMQ Support

Source: https://remoulade.readthedocs.io/en/latest/index.html

Install Remoulade with the necessary dependencies for using RabbitMQ as the message broker.

```bash
$ pip install -U 'remoulade[rabbitmq]'

```

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

### Install and Run Superbowl Dashboard

Source: https://remoulade.readthedocs.io/en/latest/getting_started.html

Clone the Superbowl repository, install its Node.js dependencies, and run the development server to access the dashboard.

```bash
$ cd ..
$ git clone https://github.com/wiremind/super-bowl.git
$ npm install
$ npm run serve

```

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

### Install Remoulade with Redis Support

Source: https://remoulade.readthedocs.io/en/latest/_sources/index.rst.txt

Installs Remoulade along with the necessary dependencies to use Redis as a message broker.

```bash
pip install -U 'remoulade[redis]'
```

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

### Instantiate RabbitmqBroker

Source: https://remoulade.readthedocs.io/en/latest/reference.html

Example of how to instantiate a RabbitmqBroker with a specific URL. Ensure RabbitMQ is running and accessible at the provided address.

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

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

### Run RabbitMQ with Docker

Source: https://remoulade.readthedocs.io/en/latest/_sources/getting_started.rst.txt

Use Docker to quickly set up a RabbitMQ message broker if you don't have it installed.

```bash
docker run -d --name rabbitmq -p 5672:5672 rabbitmq
```

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

### Run RabbitMQ with Docker

Source: https://remoulade.readthedocs.io/en/latest/getting_started.html

Use Docker to run a RabbitMQ image if you don't have it installed. This command starts a detached container named 'rabbitmq' and maps port 5672.

```bash
$ docker run -d --name rabbitmq -p 5672:5672 rabbitmq

```

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

### Initialize Prometheus Metrics and Server

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/middleware/prometheus.html

Sets up Prometheus collector registry, defines gauges and counters for various message events, and starts an HTTP server to expose metrics. This should be called before the worker starts processing messages.

```python
def before_worker_boot(self, broker, worker):
        self.logger.debug("Setting up metrics...")
        if self.registry is None:
            self.registry = prom.CollectorRegistry()
        self.worker_busy = prom.Gauge(
            "remoulade_worker_busy", "1 if the worker is processing a message, 0 if not", registry=self.registry
        )
        self.total_errored_messages = prom.Counter(
            "remoulade_message_errors_total",
            "The total number of errored messages.",
            ["queue_name", "actor_name"],
            registry=self.registry,
        )
        self.total_retried_messages = prom.Counter(
            "remoulade_message_retries_total",
            "The total number of retried messages.",
            ["queue_name", "actor_name"],
            registry=self.registry,
        )
        self.total_rejected_messages = prom.Counter(
            "remoulade_message_rejects_total",
            "The total number of dead-lettered messages.",
            ["queue_name", "actor_name"],
            registry=self.registry,
        )
        self.message_durations = prom.Summary(
            "remoulade_message_duration_milliseconds",
            "The time spent processing messages.",
            ["queue_name", "actor_name"],
            registry=self.registry,
        )
        for actor in broker.actors.values():
            self._init_labels(actor, worker)

        self.logger.debug("Starting exposition server...")
        prom.start_http_server(addr=self.http_host, port=self.http_port, registry=self.registry)
```

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

### Run Redis using Docker

Source: https://remoulade.readthedocs.io/en/latest/getting_started.html

Start a Redis instance using Docker for use with Remoulade.

```bash
$ docker run -d --name redis -p 6379:6379 redis
```

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

### Install Raven Client with Pipenv

Source: https://remoulade.readthedocs.io/en/latest/cookbook.html

Install Sentry's Raven client using Pipenv. This is necessary for integrating Sentry error reporting with Remoulade.

```bash
$ pipenv install raven

```

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

### Install Remoulade with RabbitMQ and Redis Support

Source: https://remoulade.readthedocs.io/en/latest/_sources/installation.rst.txt

Install Remoulade with support for both RabbitMQ and Redis message brokers. This command includes dependencies for both.

```bash
pip install -U 'remoulade[rabbitmq, redis]'
```

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

### Install Rollbar with Pipenv

Source: https://remoulade.readthedocs.io/en/latest/cookbook.html

Install the Rollbar Python client using Pipenv. This is a prerequisite for integrating Rollbar error reporting with Remoulade.

```bash
$ pipenv install rollbar

```

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

### Starting Time Limit Timer Thread

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/middleware/time_limit.html

Starts the background timer thread for the TimeLimit middleware if the current platform is supported (CPython). A warning is issued for unsupported platforms.

```python
def after_process_boot(self, _) -> None:
        if current_platform in supported_platforms:
            thread = threading.Thread(target=self._timer, daemon=True)
            thread.start()
        else:
            msg = f"TimeLimit cannot kill threads on your current platform {current_platform}."
            warnings.warn(msg, category=RuntimeWarning, stacklevel=2)
```

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

### Install Remoulade with RabbitMQ and Redis Support via Pip

Source: https://remoulade.readthedocs.io/en/latest/installation.html

Install Remoulade with support for both RabbitMQ and Redis message brokers by specifying them as extra requirements.

```bash
$ pip install -U ‘remoulade[rabbitmq, redis]’
```

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

### Manage Message Start Times

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/middleware/prometheus.html

Provides access to a dictionary for storing message start times, used for calculating message processing duration. Initializes the dictionary if it doesn't exist.

```python
@property
    def message_start_times(self):
        if not hasattr(self.local_data, "message_start_times"):
            self.local_data.message_start_times = {}
        return self.local_data.message_start_times
```

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

### Get State Backend

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Retrieves the StateBackend associated with the broker. Raises NoStateBackend if none is configured.

```python
def get_state_backend(self) -> StateBackend:
        """Get the StateBackend associated with the broker

        Raises:
            NoStateBackend: if there is no StateBackend

        Returns:
            StateBackend: the state backend
        """
        return self._get_backend("state")
```

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

### Set up Virtual Environment

Source: https://remoulade.readthedocs.io/en/latest/getting_started.html

Create a new directory, set up a Python virtual environment, activate it, and check the Python version. This is recommended for following the tutorial.

```bash
$ mkdir remoulade-tutorial
$ cd remoulade-tutorial
$ python3 -m venv env
$ source env/bin/activate
$ python --version

```

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

### Clone Remoulade Repository from GitHub

Source: https://remoulade.readthedocs.io/en/latest/installation.html

Clone the latest development version of Remoulade from its GitHub repository to install it from source.

```bash
$ git clone https://github.com/wiremind/remoulade
```

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

### Initialize RedisBackend with Connection Parameters

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/rate_limits/backends/redis.html

Initializes the Redis backend, optionally using a provided client, a connection URL, or Redis connection parameters. If a URL is given, it takes precedence over other parameters.

```python
from typing import Callable, List

import redis

from ..backend import RateLimiterBackend



class RedisBackend(RateLimiterBackend):
    """A rate limiter backend for Redis_.

    Parameters:
      client(Redis): An optional client.  If this is passed,
        then all other parameters are ignored.
      url(str): An optional connection URL.  If both a URL and
        connection paramters are provided, the URL is used.
      **parameters(dict): Connection parameters are passed directly
        to :class:`redis.Redis`.

    .. _redis: https://redis.io
    """

    def __init__(self, *, client=None, url=None, **parameters):
        if url is not None:
            parameters["connection_pool"] = redis.ConnectionPool.from_url(url)

        self.client = client or redis.Redis(**parameters)
```

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

### Configure Remoulade Scheduler

Source: https://remoulade.readthedocs.io/en/latest/_sources/cookbook.rst.txt

Set up a scheduler with RabbitMQ as the broker and define scheduled jobs. This example schedules the 'count_words' actor to run once every 24 hours (86400 seconds).

```python
import remoulade
from datetime import datetime
from remoulade.brokers.rabbitmq import RabbitmqBroker
from remoulade.scheduler import ScheduledJob, Scheduler

broker = RabbitmqBroker()

remoulade.set_broker(broker)
remoulade.set_scheduler(
    Scheduler(
        broker,
        [
            ScheduledJob(actor_name="count_words", interval=86400),
        ]
    )
)

broker.declare_actor(count_words)
```

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

### Pipeline Build Method

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/composition.html

Builds the pipeline by processing messages from end to start, serializing options for subsequent steps. This method is crucial for preparing the pipeline for execution.

```python
def build(self, *, last_options=None, composition_id: str = None, cancel_on_error: bool = False):
        """Build the pipeline, return the first message to be enqueued or integrated in another pipeline

        Build the pipeline by starting at the end. We build a message with all it's options in one step and
        we serialize it (asdict) as the previous message pipe_target in the next step.

        We need to know what is the options (pipe_target) of the pipeline before building it because we cannot
        edit the pipeline after it has been built.

        
```

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

### Retrieve Message Result Using Result Instance

Source: https://remoulade.readthedocs.io/en/latest/cookbook.html

This example shows how to retrieve a message result by creating a `Result` instance using the message ID. It demonstrates blocking until the result is available.

```python
import remoulade

from remoulade.brokers.rabbitmq import RabbitmqBroker
from remoulade.results.backends import RedisBackend
from remoulade.results import Results

result_backend = RedisBackend()
broker = RabbitmqBroker()
broker.add_middleware(Results(backend=result_backend))
remoulade.set_broker(broker)

@remoulade.actor(store_results=True)
def add(x, y):
  return x + y

broker.declare_actor(add)

if __name__ == "__main__":
  message = add.send(1, 2)
  result = Result(message_id=message.message_id)
  print(result.get(block=True))

```

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

### Worker Start

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/worker.html

Starts the worker's message consumption and processing sequence.

```APIDOC
## start()

### Description
Initialize the worker boot sequence and start up all the worker threads.
```

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

### Initialize Redis Backend

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/results/backends/redis.html

Initializes the Redis backend with various configuration options. Supports connection via URL or direct parameters. Defaults can be overridden by environment variables.

```python
def __init__(
    self,
    *,
    namespace="remoulade-results",
    encoder=None,
    client=None,
    url=None,
    default_timeout=None,
    max_retries=3,
    min_backoff=500,
    max_backoff=5000,
    backoff_strategy: BackoffStrategy = "spread_exponential",
    **parameters,
):
    super().__init__(namespace=namespace, encoder=encoder, default_timeout=default_timeout)

    url = url or os.getenv("REMOULADE_REDIS_URL")
    if url:
        parameters["connection_pool"] = redis.ConnectionPool.from_url(url)

    self.client = client or redis.Redis(**parameters)
    self.max_retries = max_retries
    self.min_backoff = min_backoff
    self.max_backoff = max_backoff
    self.backoff_strategy = backoff_strategy
```

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

### Initialize BucketRateLimiter

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/rate_limits/bucket.html

Instantiate a BucketRateLimiter to enforce limits on operations. Configure the maximum operations per bucket and the bucket interval in milliseconds.

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

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

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

### Initialize RabbitMQ Consumer

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/brokers/rabbitmq.html

Sets up a RabbitMQ consumer with a specified prefetch count and timeout. Handles potential connection or channel errors by raising a ConnectionClosed exception.

```python
self.logger = get_logger(__name__, type(self))
            self.channel = connection.channel()
            self.channel.basic.qos(prefetch_count=prefetch)
            self.channel.basic.consume(queue=queue_name, no_ack=False)
            self.timeout = timeout
```

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

### Initialize Results Middleware

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/results/middleware.html

Instantiate the Results middleware, providing a backend and configuring result storage options. Defaults are used if not specified.

```python
self.logger = get_logger(__name__, type(self))
self.backend = backend
self.store_results = store_results
self.result_ttl = result_ttl or DEFAULT_RESULT_TTL
```

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

### Configure Results Middleware with Redis Backend

Source: https://remoulade.readthedocs.io/en/latest/reference.html

This snippet shows how to initialize a Redis backend and add the Results middleware to the broker. Ensure Redis is running and accessible.

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

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

### Broker Initialization

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Initializes the Broker with optional middleware. If no middleware is provided, default middleware is used. It sets up logging, internal data structures for actors, queues, and middleware.

```python
def __init__(self, middleware: "Optional[Iterable[Middleware]]" = None):
        self.logger = get_logger(__name__, type(self))
        self.actors: "Dict[str, Actor]" = {}
        self.queues: Dict[str, Optional[Queue]] = {}
        self.delay_queues: Set[str] = set()

        self.actor_options: Set[str] = set()
        self.middleware: "List[Middleware]" = []
        self.group_transaction = False

        if middleware is None:
            middleware = [m() for m in default_middleware]

        for m in middleware:
            self.add_middleware(m)

        for m in extra_default_middleware:
            self.add_middleware(m)
```

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

### Start Remoulade Worker

Source: https://remoulade.readthedocs.io/en/latest/getting_started.html

Spawn Remoulade workers in a separate terminal to process enqueued tasks. This command activates the virtual environment and starts a worker for the 'get_weather' actor.

```bash
$ source env/bin/activate
$ remoulade get_weather

```

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

### Actor Initialization

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/actor.html

The `__init__` method initializes an actor with its underlying function, name, queue configuration, priority, and options.

```APIDOC
## __init__

### Description
Initializes an actor with its underlying function, name, queue configuration, priority, and options.

### Parameters
- **fn** (Callable[P, R]) - The underlying callable function for the actor.
- **actor_name** (str) - The unique name of the actor.
- **queue_name** (str) - The default queue name for this actor.
- **alternative_queues** (Optional[List[str]]) - A list of alternative queue names.
- **priority** (int) - The priority of the actor.
- **options** (Any) - Arbitrary options passed to the broker and middleware.
```

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

### Import RedisBackend for Rate Limiting

Source: https://remoulade.readthedocs.io/en/latest/reference.html

Import the RedisBackend class to use Redis for storing rate limit metadata. This is the first step in setting up a distributed rate limiter.

```python
from remoulade.rate_limits.backends import RedisBackend
```

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

### get_declared_queues

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Get all declared queues.

```APIDOC
## get_declared_queues

### Description
Get all declared queues.

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

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

### get_declared_actors

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Get all declared actors.

```APIDOC
## get_declared_actors

### Description
Get all declared actors.

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

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

### Logging Metadata with Option

Source: https://remoulade.readthedocs.io/en/latest/_sources/cookbook.rst.txt

Demonstrates adding metadata to messages using the 'logging_metadata' option.

```python
message = actor.message_with_options(logging_metadata={"id":"value"})
```

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

### get_declared_delay_queues

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Get all declared delay queues.

```APIDOC
## get_declared_delay_queues

### Description
Get all declared delay queues.

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

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

### Create RabbitMQ Virtual Host

Source: https://remoulade.readthedocs.io/en/latest/advanced.html

Use the `rabbitmqctl` command to add a virtual host for multi-tenancy. Permissions must then be set for users within that virtual host.

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

```

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

### consume

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Get an iterator that consumes messages off of the queue.

```APIDOC
## consume

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

### Parameters
#### Path 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. Defaults to 1.
- **timeout** (int) - Optional - The amount of time in milliseconds to idle for. Defaults to 30000.

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

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

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

### Initialize RedisBackend and ConcurrentRateLimiter

Source: https://remoulade.readthedocs.io/en/latest/reference.html

Instantiate a RedisBackend and then a ConcurrentRateLimiter. The ConcurrentRateLimiter is used here to create a distributed mutex with a limit of 1 operation.

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

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

### Process Message Start and End

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/middleware/prometheus.html

Records the start time of a message, sets the worker busy gauge to 1, and calculates the message duration upon completion. It also increments the error counter if an exception occurred and resets the busy gauge when no messages are being processed.

```python
def before_process_message(self, broker, message):
        self.message_start_times[message.message_id] = time.monotonic() * 1000
        self.worker_busy.set(1)
```

```python
def after_process_message(self, broker, message, *, result=None, exception=None):
        labels = self._get_labels(broker, message)
        message_start_time = self.message_start_times.pop(message.message_id, None)
        message_duration = 0
        if message_start_time is not None:
            message_duration = (time.monotonic() * 1000) - message_start_time
        self.message_durations.labels(*labels).observe(message_duration)
        if exception is not None:
            self.total_errored_messages.labels(*labels).inc()
        if not self.message_start_times:
            self.worker_busy.set(0)
```

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

### get_cancel_backend

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Get the CancelBackend associated with the broker. This backend is used for cancelling tasks.

```APIDOC
## get_cancel_backend

### Description
Get the CancelBackend associated with the broker.

### Method
GET

### Endpoint
/broker/cancel_backend

### Parameters
None

### Response
#### Success Response (200)
- **backend** (CancelBackend) - The associated cancel backend.

#### Response Example
```json
{
  "backend": "<cancel_backend_instance>"
}
```

### Raises
- **NoCancelBackend**: if there is no CancelBackend
```

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

### Configure RabbitMQ Broker

Source: https://remoulade.readthedocs.io/en/latest/guide.html

Instantiate and set the RabbitmqBroker as the global broker early in program execution. Ensure RabbitMQ is accessible at the specified URL.

```python
import remoulade

from remoulade.brokers.rabbitmq import RabbitmqBroker

rabbitmq_broker = RabbitmqBroker(url="rabbitmq")
remoulade.set_broker(rabbitmq_broker)

```

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

### Get Status

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/results/backend.html

Retrieves the count of messages from a given list that have a result stored in the backend.

```APIDOC
## get_status

### Description
Retrieves the count of messages from a given list that have a result stored in the backend.

### Method Signature
`def get_status(self, message_ids: Iterable[str]) -> int`

### Parameters
- **message_ids** (Iterable[str]) - Required - An iterable of message IDs to check the status for.

### Returns
- **int** - The number of messages that have a stored result.
```

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

### get_state_backend

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Get the StateBackend associated with the broker. This backend is used for managing task states.

```APIDOC
## get_state_backend

### Description
Get the StateBackend associated with the broker.

### Method
GET

### Endpoint
/broker/state_backend

### Parameters
None

### Response
#### Success Response (200)
- **backend** (StateBackend) - The associated state backend.

#### Response Example
```json
{
  "backend": "<state_backend_instance>"
}
```

### Raises
- **NoStateBackend**: if there is no StateBackend
```

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

### get_result_backend

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Get the ResultBackend associated with the broker. This backend is used for retrieving task results.

```APIDOC
## get_result_backend

### Description
Get the ResultBackend associated with the broker.

### Method
GET

### Endpoint
/broker/result_backend

### Parameters
None

### Response
#### Success Response (200)
- **backend** (ResultBackend) - The associated result backend.

#### Response Example
```json
{
  "backend": "<result_backend_instance>"
}
```

### Raises
- **NoResultBackend**: if there is no ResultBackend
```

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

### Configure RabbitMQ Broker with Max Priority

Source: https://remoulade.readthedocs.io/en/latest/_sources/guide.rst.txt

Instantiate a RabbitmqBroker and set the max_priority parameter. Recommended values are between 1 and 10, with a maximum of 255. Using priority values larger than max_priority will result in them being treated as max_priority.

```python
broker = RabbitmqBroker(url="rabbitmq", max_priority=PRIO_HI)
```

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

### Broker Base Class Initialization

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Base class for broker implementations. Middleware can be provided during initialization, but it's generally recommended to use `add_middleware` for configuration.

```python
class Broker:
    """Base class for broker implementations.

    Parameters:
      middleware(list[Middleware]): The set of middleware that apply
        to this broker.  If you supply this parameter, you are
        expected to declare *all* middleware.  Most of the time,
        you'll want to use :meth:`.add_middleware` instead.


```

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

### Get Pipeline Results

Source: https://remoulade.readthedocs.io/en/latest/_sources/cookbook.rst.txt

Shows how to retrieve the final result of a pipeline or iterate through intermediate results.

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

for res in pipe.results.get(block=True):
  ...
```

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

### TimeLimit Middleware Initialization

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/middleware/time_limit.html

Initializes the TimeLimit middleware with configurable time limits, checking intervals, and exit delays. The logger is also set up during initialization.

```python
def __init__(self, *, time_limit: int = 1800000, interval: int = 1000, exit_delay: Optional[int] = None) -> None:
        self.logger = get_logger(__name__, type(self))
        self.time_limit = time_limit
        self.interval = interval
        self.deadlines: "Dict[int, Optional[Tuple[float, bool, Message]]]" = {}
        self.exit_delay = exit_delay
        self.lock = threading.Lock()
```

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

### Create a Weather ETL Pipeline

Source: https://remoulade.readthedocs.io/en/latest/_sources/getting_started.rst.txt

Build a pipeline that extracts weather data for multiple cities, transforms it, and loads it using a single request. This demonstrates chaining actors and groups.

```python
from get_weather import extract_city, transform_city, load_cities
from remoulade import pipeline, group
cities = ['Paris', 'Tokyo', 'Washington', 'BrasÃlia', 'Johannesburg']
grp = group([extract_city.message(city) | transform_city.message() for city in cities])
weather_etl_pipeline = grp | load_cities.message()
weather_etl_pipeline.run()
```

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

### Get Status of Messages

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/results/backend.html

Counts the number of messages from a given iterable that have a result stored in the backend.

```python
def get_status(self, message_ids: Iterable[str]) -> int:
        """Given a list of messages ids return the number of messages with a result stored"""
        count = 0
        for message_id in message_ids:
            message_key = self.build_message_key(message_id)
            result = self._get(message_key, forget=False)
            if result != Missing:
                count += 1
        return count
```

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

### Get Last Pipeline Result

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/composition.html

Retrieves the result of the last message or group within the pipeline.

```python
@property
def result(self) -> ResultsT:
        """Result of the last message/group of the pipeline"""
        last_child = self.children[-1]
        return cast(ResultsT, last_child.results if isinstance(last_child, group) else last_child.result)
```

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

### Get Pipeline Results

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/composition.html

Aggregates the results from all children (messages or groups) in the pipeline into a CollectionResults object.

```python
@property
def results(self) -> CollectionResults[Any]:
        """CollectionResults created from this pipeline, used for result related methods"""
        results: List[Union[Result, CollectionResults]] = []
        for element in self.children:
            results += [element.results if isinstance(element, group) else element.result]
        return CollectionResults(results)
```

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

### Initialize WorkerThread

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/worker.html

Sets up the WorkerThread with broker, consumers, work queue, and timeout settings. It runs messages off the work queue.

```python
def __init__(self, *, broker, consumers, work_queue, worker_timeout):
        super().__init__(daemon=True)

        self.logger = get_logger(__name__, "WorkerThread")
        self.running = False
        self.paused = False
        self.paused_event = Event()
        self.broker = broker
        self.consumers = consumers
        self.work_queue = work_queue
        self.timeout = worker_timeout / 1000
```

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

### Add Virtual Host and Set Permissions (RabbitMQ)

Source: https://remoulade.readthedocs.io/en/latest/_sources/advanced.rst.txt

Use rabbitmqctl to add a virtual host and set permissions for a user. This is a prerequisite for multi-tenancy in RabbitMQ.

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

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

### Get Cancel Backend

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Retrieves the CancelBackend associated with the broker. Raises NoCancelBackend if none is configured.

```python
def get_cancel_backend(self) -> CancelBackend:
        """Get the CancelBackend associated with the broker

        Raises:
            NoCancelBackend: if there is no CancelBackend

        Returns:
            CancelBackend: the cancel backend
        """
        return self._get_backend("cancel")
```

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

### Instantiate BucketRateLimiter for 1 Operation/Minute

Source: https://remoulade.readthedocs.io/en/latest/reference.html

Configure a BucketRateLimiter to permit only 1 operation within a 60000ms (1 minute) bucket. This is useful for enforcing slower, periodic operation limits.

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

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

### Set Global RabbitMQ Broker

Source: https://remoulade.readthedocs.io/en/latest/_sources/guide.rst.txt

Configure the RabbitMQ host by instantiating a RabbitmqBroker and setting it as the global broker early in your program's execution. Ensure necessary imports are included.

```python
import remoulade

from remoulade.brokers.rabbitmq import RabbitmqBroker

rabbitmq_broker = RabbitmqBroker(url="rabbitmq")
remoulade.set_broker(rabbitmq_broker)
```

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

### Get Result Backend

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Retrieves the ResultBackend associated with the broker. Raises NoResultBackend if none is configured.

```python
def get_result_backend(self) -> ResultBackend:
        """Get the ResultBackend associated with the broker

        Raises:
            NoResultBackend: if there is no ResultBackend

        Returns:
            ResultBackend: the result backend
        """
        return self._get_backend("results")
```

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

### Initialize ResultBackend

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/results/backend.html

Initializes the ResultBackend with a namespace, encoder, and default timeout. Uses a default JSON encoder if none is provided.

```python
def __init__(self, *, namespace: str = "remoulade-results", encoder: Encoder = None, default_timeout: int = None):
    from ..message import get_encoder

    self.namespace = namespace
    self.encoder = encoder or get_encoder()
    self.default_timeout = default_timeout or DEFAULT_TIMEOUT
```

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

### Worker Initialization

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/worker.html

Initializes a Worker instance. It sets up logging, connects to the broker, and configures consumer and worker threads. Raises an error if a LocalBroker is used.

```python
class Worker:
    """Workers consume messages off of all declared queues and
    distribute those messages to individual worker threads for
    processing.  Workers don't block the current thread so it's
    up to the caller to keep it alive.

    Don't run more than one Worker per process.

    Parameters:
      broker(Broker)
      queues(Set[str]): An optional subset of queues to listen on.  By
        default, if this is not provided, the worker will listen on
        all declared queues.
      worker_timeout(int): The number of milliseconds workers should
        wake up after if the queue is idle.
      worker_threads(int): The number of worker threads to spawn.
      prefetch_multiplier(int): The number of message to prefetch at a time, to be multiplied with the number of threads
    """

    def __init__(self, broker, *, queues=None, worker_timeout=1000, worker_threads=8, prefetch_multiplier=2):
        self.logger = get_logger(__name__, type(self))
        self.broker = broker
        if broker.local:
            raise RemouladeError("LocalBroker is not destined to be used with a Worker")

        self.consumers: Dict[str, _ConsumerThread] = {}
        if queues is None or len(queues) != 1:
            self.logger.warning(
                "You are listening on multiple queues, RMQ is adapted to single queue consuming only. "
                "It will work but you should consider using one worker per queue."
            )
        self.consumer_whitelist = queues and set(queues)
        # Load a small factor more messages than there are workers to
        # avoid waiting on network IO as much as possible.  The factor
        # must be small so we don't starve other workers out.
        self.queue_prefetch = min(worker_threads * max(prefetch_multiplier, 1), 65535)
        # Load a large factor more delay messages than there are
        # workers as those messages could have far-future etas.
        self.delay_prefetch = min(worker_threads * 1000, 65535)

        self.workers = []  # type: List
        self.work_queue = PriorityQueue()  # type: PriorityQueue
        self.worker_timeout = worker_timeout
        self.worker_threads = worker_threads

```

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

### Get Result of a Single Actor

Source: https://remoulade.readthedocs.io/en/latest/getting_started.html

Send a message to an actor and retrieve its result, blocking until the result is available.

```python
>>> from get_weather import extract_city
>>> message = extract_city.send("Paris")
>>> result = message.result.get(block=True)
>>> print(result)

```

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

### Get Group Message IDs

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/results/backend.html

Retrieves the collection of message IDs associated with a given group ID.

```APIDOC
## get_group_message_ids

### Description
Retrieves the collection of message IDs associated with a given group ID.

### Method Signature
`def get_group_message_ids(self, group_id: str) -> Iterable[str]`

### Parameters
- **group_id** (str) - Required - The ID of the group.

### Returns
- **Iterable[str]** - An iterable of message IDs belonging to the group.

### Raises
- **MessageIdsMissing** - If no message IDs are found for the specified group.
```

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

### Build a Message with Custom Options

Source: https://remoulade.readthedocs.io/en/latest/reference.html

Create a message with specific options, such as a custom queue name or other broker/middleware specific settings, using .message_with_options().

```python
>>> add.message_with_options(queue_name='high_priority', priority=5)
Message(queue_name='high_priority', actor_name='add', args=(), kwargs={}, options={'priority': 5}, message_id='...', message_timestamp=...)
```

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

### Instantiate BucketRateLimiter for 10 Operations/Second

Source: https://remoulade.readthedocs.io/en/latest/reference.html

Create a BucketRateLimiter instance to allow up to 10 operations within a 1000ms (1 second) bucket. This is a common configuration for high-frequency, short-term rate limiting.

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

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

### Encoder Management

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/message.html

Functions to get and set the global encoder instance used for serializing and deserializing messages.

```APIDOC
## get_encoder

### Description
Get the global encoder object.

### Returns
- Encoder: The current global encoder instance.

## set_encoder

### Description
Set the global encoder object.

### Parameters
- **encoder** (Encoder) - The encoder instance to use when serializing messages.
```

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

### Get Global Encoder

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/message.html

Retrieves the globally configured encoder instance. This is used for serializing and deserializing messages.

```python
def get_encoder() -> Encoder:
    """Get the global encoder object.

    Returns:
      Encoder
    """
    global global_encoder
    return global_encoder
```

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

### Prometheus Middleware Initialization

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/middleware/prometheus.html

Initializes the Prometheus middleware with configurable HTTP host, port, and registry. Defaults can be overridden by environment variables.

```python
from remoulade.middleware import Prometheus

prometheus_middleware = Prometheus(
    http_host="0.0.0.0",
    http_port=9090,
    registry=my_custom_registry
)
```

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

### Logging Metadata with Getter Callback

Source: https://remoulade.readthedocs.io/en/latest/_sources/cookbook.rst.txt

Shows how to dynamically provide metadata for logging by using the 'logging_metadata_getter' option with a callback function.

```python
def callback():
    return {"id":"value"}

message = actor.message_with_options(logging_metadata_getter=callback)
```

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

### Get Channel with Timeout

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/brokers/rabbitmq.html

Retrieves a channel from the pool, waiting up to a specified timeout. Raises ChannelPoolTimeout if the timeout is reached.

```python
def get(self, *, timeout=None):
        try:
            return self._pool.get(timeout=timeout)
        except Empty as e:
            raise ChannelPoolTimeout("Could not get any channel from the pool") from e
```

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

### Get Global Broker Instance

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/broker.html

Retrieves the globally configured broker instance. Raises a ValueError if no broker has been set.

```python
def get_broker() -> "Broker":
    """Get the global broker instance.  If no global broker is set,
    this initializes a RabbitmqBroker and returns it.

    Returns:
      Broker: The default Broker.
    Raises:
      ValueError if no broker is set
    """
    global global_broker
    if global_broker is None:
        raise ValueError("Broker not found, are you sure you called set_broker(broker) ?")
    return global_broker
```

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

### Initialize Prometheus Metrics Labels

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/middleware/prometheus.html

Initializes Prometheus metrics for all configured labels, ensuring that metrics are set up for each queue and actor combination.

```python
def _init_labels(self, actor, worker=None):
    # initialize the metrics for all labels to 0
    metrics = [
        self.total_errored_messages,
        self.total_retried_messages,
        self.total_rejected_messages,
        self.message_durations,
    ]
    worker_queues = worker.consumer_whitelist if worker else None
    for metric in metrics:
        if metric:  # metric can be None if actor is declared before worker boot
            label = self._get_label(actor)

            queues = set(actor.queue_names)
            if worker_queues:
                queues &= worker_queues

            for queue_name in queues:
                metric.labels(queue_name, label)
```

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

### Actor Class Initialization

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/actor.html

Initializes an Actor instance with its associated function, name, queue configuration, priority, and options. It also sets up a logger for the actor.

```python
def __init__(
        self,
        fn: Callable[P, R],
        *, 
        actor_name: str,
        queue_name: str,
        alternative_queues: Optional[List[str]],
        priority: int,
        options: Any,
    ) -> None:
        self.logger = get_logger(fn.__module__, actor_name)
        self.fn = fn
        self.broker: "Optional[Broker]" = None
        self.actor_name = actor_name
        self.queue_name = queue_name
        self.alternative_queues = alternative_queues
        self.priority = priority
        self.options = options
```

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

### Get Prometheus Labels for Message

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/middleware/prometheus.html

Retrieves the queue name and the Prometheus label for a given message and its associated actor.

```python
def _get_labels(self, broker, message):
    actor = broker.get_actor(message.actor_name)
    return message.queue_name, self._get_label(actor)
```

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

### Initialize Retries Middleware

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/middleware/retries.html

Instantiate the Retries middleware with custom or default backoff settings and retry conditions.

```python
Retries(
    max_retries=5,
    min_backoff=10000,
    max_backoff=60000,
    retry_when=lambda retries, exc: isinstance(exc, ConnectionError),
    backoff_strategy="exponential",
    jitter=True,
    increase_priority_on_retry=True
)
```

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

### Internal Get Method

Source: https://remoulade.readthedocs.io/en/latest/_modules/remoulade/rate_limits/backends/stub.html

Retrieves the value of a key if it exists and has not expired. Returns a default value if the key is not found or has expired.

```python
backend._get('user:123', default=0)
```

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

### Configure Local Broker for Development

Source: https://remoulade.readthedocs.io/en/latest/_sources/guide.rst.txt

Set up a LocalBroker for development environments to execute actors when messages are enqueued without running a separate Worker. This configuration includes setting up Results and Cancel middleware with local backends.

```python
import remoulade

from remoulade.brokers.local import LocalBroker
from remoulade.results.backends import LocalBackend
from remoulade.cancel.backends import StubBackend

local_broker = LocalBroker(middleware=[])
broker.add_middleware(Results(backend=LocalBackend()))
broker.add_middleware(Cancel(backend=StubBackend()))
remoulade.set_broker(local_broker)
```
Tokens:
Raw