Remoulade

https://github.com/wiremind/remoulade
Admin
Remoulade is a fast and reliable distributed task processing library for Python 3, forked from...
Tokens:34,504
Snippets:517
Trust Score:9
Update:3 months ago
Show doc for...
Tokens:
Raw
### Install Remoulade from Source

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/installation.md

Install Remoulade locally after cloning the repository by running the pip install command in the cloned directory.

```bash
$ python -m pip install .
```

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

### Install Remoulade with All Features

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/installation.md

Install Remoulade with all available features and dependencies using pip.

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

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

### Install and Run Superbowl Dashboard

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

Commands to clone the Superbowl repository, install dependencies, and start the dashboard server. This is used for monitoring Remoulade tasks.

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

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

### Install Remoulade with Redis Support

Source: https://github.com/wiremind/remoulade/blob/master/README.md

Use this command to install Remoulade with Redis integration. Ensure you have uv and pip installed.

```console
uv pip install 'remoulade[redis]'
```

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

### Install Remoulade with Redis Support

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/index.md

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

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

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

### Install Remoulade with RabbitMQ Support

Source: https://github.com/wiremind/remoulade/blob/master/README.md

Use this command to install Remoulade with RabbitMQ integration. Ensure you have uv and pip installed.

```console
uv pip install 'remoulade[rabbitmq]'
```

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

### Run Redis with Docker

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

Start a Redis instance using Docker for Remoulade to connect to.

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

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

### Install Remoulade with RabbitMQ and Requests

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

Installs Remoulade with RabbitMQ support and the Requests library. This command ensures you have the necessary dependencies for the tutorial.

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

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

### Install Remoulade with RabbitMQ Support

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/index.md

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

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

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

### Run Test Suite with Docker Compose

Source: https://github.com/wiremind/remoulade/blob/master/CONTRIBUTING.md

To run the test suite, you need RabbitMQ and Redis. This command starts these services using Docker Compose. Ensure you are in the 'tests/' directory before execution.

```bash
cd tests/
docker compose up
```

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

### Basic Remoulade Actor Example

Source: https://github.com/wiremind/remoulade/blob/master/README.md

This Python script demonstrates setting up a RabbitMQ broker, defining a simple actor to count words from a URL, and sending tasks to the broker. Ensure RabbitMQ is running before execution.

```python
from remoulade.brokers.rabbitmq import RabbitmqBroker
import remoulade
import requests
import sys

broker = RabbitmqBroker()
remoulade.set_broker(broker)


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


broker.declare_actor(count_words)

if __name__ == "__main__":
    count_words.send(sys.argv[1])
```

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

### Install Rollbar Client

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Install the Rollbar Python client using pipenv. This is a prerequisite for integrating Rollbar error reporting with your project.

```bash
$ pipenv install rollbar
```

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

### Install Remoulade with RabbitMQ and Redis Support

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/installation.md

Install Remoulade with support for both RabbitMQ and Redis message brokers using pip.

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

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

### Create a Task Pipeline with pipeline

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

Chain actors sequentially using `remoulade.pipeline`. This example demonstrates creating a weather ETL pipeline for a single city.

```default
>>> from remoulade import pipeline
>>> from get_weather import extract_city, transform_city, load_city
>>> weather_etl_pipeline = pipeline([extract_city.message("Paris"), transform_city.message(), load_city.message()])
>>> weather_etl_pipeline.run()
```

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

### Install Sentry Raven Client

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Install the Sentry raven Python client using pipenv. This is required for integrating Sentry error reporting.

```bash
$ pipenv install raven
```

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

### Run RabbitMQ with Docker

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

Starts a RabbitMQ broker in detached mode using Docker. This is a prerequisite for using Remoulade with RabbitMQ.

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

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

### Display Progress for Remoulade Group with tqdm

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Integrates the tqdm library to display a progress bar for a Remoulade group's execution. This is useful for long-running operations where visual feedback is desired. Ensure tqdm is installed (`pip install tqdm`).

```python
from time import time, sleep
import logging

from tqdm import tqdm
import remoulade

logger = logging.getLogger(__name__)

def blocking_remoulade_group(remoulade_group, timeout=1800):
    actor_count = len(remoulade_group)
    logger.info('Start group')
    start_time = time()
    try:
        with tqdm(total=actor_count) as progress_bar:
            results = remoulade_group.run().results
            completed_count, waited_time = 0, 0
            while waited_time < timeout and completed_count != actor_count:
                completed_count = results.completed_count
                waited_time = time() - start_time
                progress_bar.update(completed_count - progress_bar.n)
                sleep(1)

            progress_bar.update(completed_count - progress_bar.n)  # final update of the progress

        if waited_time > timeout:
            raise Exception('The operation timed out')

    except:
        logger.error('Group canceled ')
        remoulade_group.cancel()
        raise

    logger.info('Finished group')

    return results
```

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

### Run Remoulade Workers

Source: https://github.com/wiremind/remoulade/blob/master/README.md

Execute this command in a terminal to start Remoulade workers that will process tasks from the broker. This command assumes your task definitions are in a file named 'example.py'.

```console
remoulade example
```

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

### Conditional Broker Setup for Unit Tests

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Conditionally set up either a StubBroker for unit tests (when UNIT_TESTS environment variable is '1') or a RabbitmqBroker otherwise. This allows testing without a running RabbitMQ instance.

```python
import os

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

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

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

### Start Remoulade Workers

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

This command starts Remoulade workers in a separate terminal to process messages from the broker. It requires activating the virtual environment and running the `remoulade` command with the actor name.

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

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

### Clone Remoulade Repository

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/installation.md

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

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

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

### Remoulade Worker Boot-up Logs

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Observe these log messages to confirm that Remoulade workers have successfully started and are ready to process tasks. The logs indicate 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.
```

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

### Run Synchronous Word Count Function

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Example of calling the synchronous `count_words` function in an interactive Python session.

```python
>>> from count_words import count_words
>>> count_words("http://example.com")
There are 338 words at 'http://example.com'.
```

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

### Define a Python Function to Fetch Weather

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

This function fetches weather data for a given city and posts the description to a specified endpoint. Ensure the `requests` library is installed.

```python
import requests

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

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

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

### Test Actor Execution with Stub Broker and Worker

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Example test function demonstrating how to use stub_broker and stub_worker fixtures to send a message and ensure it's processed. It joins the broker and worker to wait for completion.

```default
def test_count_words(stub_broker, stub_worker):
  count_words.send("http://example.com")
  stub_broker.join(count_words.queue_name)
  stub_worker.join()
```

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

### Send a Message to a Remoulade Actor

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

This example shows how to asynchronously send a message to the `get_weather` actor using its `send` method. This enqueues a message to the configured broker (RabbitMQ) for processing by a worker.

```default
>>> from get_weather import get_weather
>>> get_weather.send("Lyon")
Message(queue_name='default', actor_name='get_weather', args=('Lyon',), kwargs={}, options={}, message_id='686f9577-d5d9-4853-a2fb-66bde2e60098', message_timestamp=1625665996101)
```

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

### Create and Get a Result from Message ID

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Demonstrates how to create a `Result` object directly from a message ID and retrieve its value. This is an alternative to accessing the result via the `Message` object.

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

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

### Store and Retrieve Message Results with Redis

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Configure Remoulade to store message results using a Redis backend. This example demonstrates setting up the broker with the Results middleware and sending a message, then retrieving its result.

```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)
  print(message.result.get(block=True, raise_on_error=True, forget=False))
```

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

### Spawn Remoulade Workers

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Use this command to start Remoulade worker processes for a specific actor function. This command spawns a process with 8 worker threads by default.

```bash
$ remoulade count_words
```

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

### Declare a Remoulade Actor with RabbitMQ Broker

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

This code snippet shows how to define a Remoulade actor using the `@remoulade.actor` decorator and configure it to use RabbitMQ as the message broker. Ensure `remoulade` and `pika` (for RabbitMQ) are installed.

```python
import requests
import remoulade
from remoulade.brokers.rabbitmq import RabbitmqBroker

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)
remoulade.declare_actors([get_weather])
```

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

### Benchmark Help Options

Source: https://github.com/wiremind/remoulade/blob/master/benchmarks/README.md

View all available options for running the benchmarks.

```bash
python bench.py --help
```

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

### Run Remoulade Benchmark

Source: https://github.com/wiremind/remoulade/blob/master/benchmarks/README.md

Execute the Remoulade benchmark. Prepend `env REDIS=1` to use Redis as the backend.

```bash
python bench.py
```

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

### Broker Class Initialization

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Initializes a new Broker instance. Middleware can be provided during initialization, but it's generally recommended to use `add_middleware` for managing middleware.

```APIDOC
## Broker(middleware: Iterable[[Middleware](#remoulade.Middleware)] | None = None)

### Description
Base class for broker implementations.

### Parameters:
- **middleware** (*list* *[*[*Middleware*](#remoulade.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 [`add_middleware()`](#remoulade.Broker.add_middleware) instead.
```

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

### Configure Redis Backend for Actor Results

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Set up the Redis backend and add the Results middleware to the broker. Ensure Redis is accessible.

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

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

### Run Celery Benchmark

Source: https://github.com/wiremind/remoulade/blob/master/benchmarks/README.md

Execute the Celery benchmark. Prepend `env REDIS=1` to use Redis as the backend.

```bash
python bench.py --use-celery
```

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

### Configure RabbitMQ Broker

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Instantiate and set the RabbitmqBroker as the global broker. This should be done early in your program's execution.

```default
import remoulade

from remoulade.brokers.rabbitmq import RabbitmqBroker

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

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

### Configure Local Broker for Development

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Set up a LocalBroker for development environments to execute actors immediately upon message enqueueing, without requiring a separate Worker process. It integrates with LocalBackend for results and StubBackend for cancellation.

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

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

### Send Actor Message and Get Result

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

Send a message to an actor and then block until its result is available using the .result.get() method.

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

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

### Declare an Actor

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Use the remoulade.actor decorator to declare a function as an actor. This example shows a simple 'add' function. You must declare actors before using them.

```python
@remoulade.actor
def add(x, y):
  print(x + y)

add
```

```python
remoulade.declare_actors([add])
```

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

### Create and Pipe Messages

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Demonstrates how to create a message with arguments and pipe it to another message, resulting in a pipeline of operations.

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

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

### Add RabbitMQ Virtual Host and Permissions

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/advanced.md

Use `rabbitmqctl` to create a new virtual host and set permissions for a user. This is necessary for multi-tenancy configurations in RabbitMQ.

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

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

### Import Remoulade

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Import the Remoulade library to begin using its functionalities.

```python
import remoulade
```

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

### Build Message with Options

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Constructs a message with arbitrary processing options, useful for composing actors. Ensure all required arguments are provided.

```python
message_with_options(args=(1,), kwargs={'a': 1}, queue_name='my_queue', timeout=10)
```

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

### Send Message with Options

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Asynchronously sends a message with additional options for the broker and middleware, including a delay.

```python
send_with_options(args=(1,), kwargs={'a': 1}, queue_name='my_queue', delay=100, timeout=10)
```

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

### Use Callback for Logging Metadata

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Provide a callback function to dynamically generate metadata for logging. The callback should return a dictionary of metadata. This method is an alternative to directly providing logging_metadata.

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

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

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

### Pytest Fixtures for Stub Broker and Worker

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Define pytest fixtures for a stub broker and a worker. The stub_broker fixture flushes all data before each test, and the stub_worker fixture starts and stops a Worker instance for testing purposes.

```python
import remoulade
import pytest

from remoulade 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()
```

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

### Run a Python Function in the Shell

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

This demonstrates how to import and execute the `get_weather` function from a Python shell. It requires the `get_weather.py` file to be in the current directory or Python path.

```default
$ python
>>> from get_weather import get_weather
>>> get_weather("Paris")
```

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

### Actor Typing with PEP 612

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/changelog.md

Demonstrates the new actor typing using PEP 612 parameter specification variables. Previously, 'add' was Actor[Callable[[int, int], int]], now it's Actor[[int, int], int].

```python
from typing import Callable
from remoulade import actor

@actor
def add(a: int, b: int) -> int:
   return a + b

# previously `add` was Actor[Callable[[int, int], int]]
# now it's Actor[[int, int], int]
# and Actor[Callable[..., int]] becomes Actor[..., int]
```

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

### Initialize and Add Rollbar Middleware to Broker

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Initialize the Rollbar client with your API key and add the custom Rollbar middleware to your Remoulade broker. This enables error reporting.

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

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

### Instantiate and Add Sentry Middleware to Broker

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Create an instance of the Sentry raven client with your DSN and add the Sentry middleware to your Remoulade broker. This activates Sentry error reporting.

```python
from raven import Client

raven_client = Client(YOUR_DSN)
broker.add_middleware(path.to.SentryMiddleware(raven_client))
```

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

### Create and Run an Actor Pipeline

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

Construct and execute a pipeline that chains multiple actors, including a group of actors feeding into a single actor, to process data in stages.

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

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

### Create a Task Pipeline with Pipe Notation

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

Chain actors sequentially using the pipe operator (`|`). This provides a more readable syntax for creating task pipelines, similar to the `pipeline` function.

```default
>>> from get_weather import extract_city, transform_city, load_city
>>> weather_etl_pipeline = extract_city.message("Paris") | transform_city.message() | load_city.message()
>>> weather_etl_pipeline.run()
```

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

### Send Actor Tasks

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Demonstrates sending instances of `FooTask` and `BarTask` for execution.

```python
>>> FooTask.send()
>>> BarTask.send()
```

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

### Broker.get_state_backend

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

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

```APIDOC
## Broker.get_state_backend() → StateBackend

### Description
Get the StateBackend associated with the broker

### Raises:
**NoStateBackend** – if there is no StateBackend

### Returns:
the state backend

### Return type:
StateBackend
```

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

### RabbitmqBroker with Max Priority

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Configure RabbitmqBroker with `max_priority` to leverage RabbitMQ's priority queue feature. Use constants for priority levels.

```python
PRIO_LO = 0
PRIO_MED = 1
PRIO_HI = 2
broker = RabbitmqBroker(url=”rabbitmq”, max_priority=PRIO_HI)
```

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

### Schedule Weather Updates Every 10 Seconds

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

This snippet demonstrates how to configure Remoulade to periodically update weather data. Ensure you replace '<unique-id>' with your actual endpoint.

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

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

### Enqueue Tasks to Remoulade

Source: https://github.com/wiremind/remoulade/blob/master/README.md

Run this command in a separate terminal to send tasks to the Remoulade broker. Replace the URL with the target website.

```console
python3 example.py http://example.com
```

```console
python3 example.py https://github.com
```

```console
python3 example.py https://news.ycombinator.com
```

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

### Prepare Directory for Prometheus Metrics

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/advanced.md

When using Prometheus with Remoulade in a multi-process configuration, ensure the prometheus_multiproc_dir and remoulade_prom_db environment variables point to an existing folder. Clean the folder before running Remoulade to avoid metric export issues.

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

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

### Configure Prometheus Metrics Export

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/advanced.md

Set environment variables to specify the host and port for the Prometheus exposition server. This is automatically enabled when running workers via the command line utility.

```bash
export remoulade_prom_host=<host>
export remoulade_prom_port=<port>
```

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

### remoulade.Message.build

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Build message for pipeline

```APIDOC
## build

### Description
Build message for pipeline

### Parameters
#### Parameters
- **options** (dict) - Options for building the message.
```

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

### Broker.get_cancel_backend

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

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

```APIDOC
## Broker.get_cancel_backend() → CancelBackend

### Description
Get the CancelBackend associated with the broker

### Raises:
**NoCancelBackend** – if there is no CancelBackend

### Returns:
the cancel backend

### Return type:
CancelBackend
```

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

### remoulade.get_broker()

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Retrieves the global broker instance. If none is set, it initializes and returns a RabbitmqBroker.

```APIDOC
## Function: remoulade.get_broker()

### Description
Get the global broker instance. If no global broker is set, this initializes a RabbitmqBroker and returns it.

### Returns
The default Broker.

### Return Type
[Broker](#remoulade.Broker)

### Raises
ValueError if no broker is set
```

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

### Schedule a Message with Delay

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Use `send_with_options` with a `delay` argument to schedule a message to be processed after a specified time in milliseconds.

```python
>>> count_words.send_with_options(args=("https://example.com",), delay=10000)
Message(
  queue_name='default',
  actor_name='count_words',
  args=('https://example.com',), kwargs={},
  options={'eta': 1498560453548},
  message_id='7387dc76-8ebe-426e-aec1-db34c236563c',
  message_timestamp=1498560443548)
```

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

### Implement Rate Limiting with Middleware

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Enforce rate limits on message enqueuing and processing using `RateLimitEnqueue` and `RateLimitProcess` middleware with a Redis backend. Configure limits directly on actors and handle `RateLimitExceeded` exceptions.

```python
import remoulade

from remoulade.brokers.rabbitmq import RabbitmqBroker
from remoulade.rate_limits import RateLimitEnqueue, RateLimitProcess
from remoulade.rate_limits.backends import RedisBackend

broker = RabbitmqBroker()
backend = RedisBackend()
broker.add_middleware(RateLimitEnqueue(backend=backend))
broker.add_middleware(RateLimitProcess(backend=backend))

@remoulade.actor(enqueue_rate_limits="2 per second", process_rate_limits="5 per minute")
def limited():
  return "ok"

broker.declare_actor(limited)

limited.send()
limited.send()

try:
  limited.send()
except remoulade.RateLimitExceeded:
  print("Too many requests")
```

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

### Create Sentry Error Reporting Middleware

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Implement a Remoulade middleware to capture and report exceptions using Sentry's raven client. Ensure the middleware is initialized with a configured raven client.

```python
import remoulade

class SentryMiddleware(remoulade.Middleware):
  def __init__(self, raven_client):
    self.raven_client = raven_client

  def after_process_message(self, broker, message, *, result=None, exception=None):
    if exception is not None:
      self.raven_client.captureException()
```

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

### Implement Actor Task with Overridden Meta

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Creates a concrete actor task 'BarTask' inheriting from `BaseTask`, overriding meta options like `max_retries`, and implementing `get_task_name`.

```python
>>> class BarTask(BaseTask):
...   class Meta(BaseTask.Meta):
...     max_retries = 10
...
...   def get_task_name(self):
...     return "Bar"
```

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

### Chain Actors with Remoulade Pipelines

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Combine actors sequentially using pipelines, where the output of one actor becomes the input for the next. The result of the last actor can be retrieved using `.result.get()`.

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

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

```default
uri = "http://example.com"
pipe = pipeline([
  get_uri_contents.message(uri),
  count_words.message(uri),
]).run()
```

```default
pipe = get_uri_contents.message(uri) | count_words.message(uri)
```

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

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

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

### Build Message for Pipeline

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Builds a message specifically for use within a pipeline, accepting a dictionary of options.

```python
msg.build(options={'timeout': 10})
```

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

### Define Base Actor with Meta Options

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Defines a base actor class with abstract meta options for queue name and retries. Subclasses must implement `get_task_name` and `perform`.

```python
>>> class BaseTask(GenericActor):
...   class Meta:
...     abstract = True
...     queue_name = "tasks"
...     max_retries = 20
...
...   def get_task_name(self):
...     raise NotImplementedError
...
...   def perform(self):
...     print(f"Hello from {self.get_task_name()}!")
```

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

### Configure Remoulade with Results Middleware and Redis Backend

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/getting_started.md

Set up Remoulade to use the Results middleware with a Redis backend for storing actor results. Ensure actors are declared after configuration.

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


@remoulade.actor
def extract_city(city):
   ...


@remoulade.actor()
def transform_city(response):
    ...


@remoulade.actor
def load_city(text):
   ...


result_backend = RedisBackend()
rabbitmq_broker = RabbitmqBroker()
result_time_limit_ms = 10 * 60 * 1000 # 10 mn
rabbitmq_broker.add_middleware(Results(backend=result_backend, store_results=True, result_ttl=result_time_limit_ms))
remoulade.set_broker(rabbitmq_broker)
remoulade.declare_actors([extract_city, transform_city, load_city])
```

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

### Broker.get_result_backend

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

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

```APIDOC
## Broker.get_result_backend() → [ResultBackend](#remoulade.results.ResultBackend)

### Description
Get the ResultBackend associated with the broker

### Raises:
[**NoResultBackend**](#remoulade.NoResultBackend) – if there is no ResultBackend

### Returns:
the result backend

### Return type:
[ResultBackend](#remoulade.results.ResultBackend)
```

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

### Enqueue Message on Redis List

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/advanced.md

Use Redis commands `HSET` and `RPUSH` to enqueue a message. The message payload is stored as a hash field, and its ID is pushed to a list representing the queue.

```bash
> HSET default.msgs $YOUR_MESSAGE_ID $YOUR_MESSAGE_PAYLOAD
> RPUSH default $YOUR_MESSAGE_ID
```

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

### Configure Constant Backoff Strategy for an Actor

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Set a fixed delay between message retries using the `constant` backoff strategy. Specify the delay in milliseconds using `min_backoff` and set `backoff_strategy` to 'constant'.

```python
import remoulade

@remoulade.actor(min_backoff=60000, backoff_strategy='constant')
def count_words(URL):
  ...
```

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

### remoulade.Message.message_with_options

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Build a message with an arbitrary set of processing options. This method is useful if you want to compose actors.

```APIDOC
## message_with_options

### Description
Build a message with an arbitrary set of processing options. This method is useful if you want to compose actors.

### Parameters
#### Parameters
- **args** (tuple) - Positional arguments that are passed to the actor.
- **kwargs** (dict) - Keyword arguments that are passed to the actor.
- **queue_name** (str) - Name of the queue to put this message into when enqueued.
- **\*\*options** (dict) - Arbitrary options that are passed to the broker and any registered middleware.

### Returns
- A message that can be enqueued on a broker.
- **Return type:** [Message](#remoulade.Message)
```

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

### remoulade.middleware.MiddlewareError

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Base class for all middleware-related errors in Remoulade.

```APIDOC
### class remoulade.middleware.MiddlewareError

Base class for middleware errors.
```

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

### remoulade.set_broker(broker)

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Configures the global broker instance to be used by default.

```APIDOC
## Function: remoulade.set_broker(broker)

### Description
Configure the global broker instance.

### Parameters
* **broker** ([*Broker*](#remoulade.Broker)) – The broker instance to use by default.

### Returns
None
```

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

### Consumer Class Methods

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

The Consumer class provides methods to interact with message queues. It allows iteration over messages, acknowledgment, and rejection of messages.

```APIDOC
## class remoulade.Consumer

Consumers iterate over messages on a queue.

Consumers and their MessageProxies are *not* thread-safe.

### __iter__()

Returns this instance as a Message iterator.

### __next__()

Retrieve the next message off of the queue. This method
blocks until a message becomes available.

* **Returns:**
  A transparent proxy around a Message that can
  be used to acknowledge or reject it once it’s done being
  processed.
* **Return type:**
  [MessageProxy](#remoulade.MessageProxy)

### ack(message)

Acknowledge that a message has been processed, removing it
from the broker.

* **Parameters:**
  **message** ([*MessageProxy*](#remoulade.MessageProxy)) – The message to acknowledge.

### close()

Close this consumer and perform any necessary cleanup actions.

### nack(message)

Move a message to the dead-letter queue.

* **Parameters:**
  **message** ([*MessageProxy*](#remoulade.MessageProxy)) – The message to reject.
```

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

### Broker.enqueue

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Enqueues a message onto the broker, with an optional delay. Returns the enqueued message.

```APIDOC
## Broker.enqueue(message: [Message](#remoulade.Message)[Any], , delay: int | None = None) → [Message](#remoulade.Message)[Any]

### Description
Enqueue a message on this broker.

### Parameters:
- **message** ([*Message*](#remoulade.Message)) – The message to enqueue.
- **delay** (*int*) – The number of milliseconds to delay the message for.

### Returns:
Either the original message or a copy of it.

### Return type:
[Message](#remoulade.Message)
```

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

### remoulade.Message.copy

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Create a copy of this message.

```APIDOC
## copy

### Description
Create a copy of this message.

### Parameters
#### Parameters
- **\*\*attributes** - Attributes to copy.
```

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

### Implement Specific Actor Task

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Creates a concrete actor task 'FooTask' by inheriting from `BaseTask` and implementing `get_task_name`.

```python
>>> class FooTask(BaseTask):
...   def get_task_name(self):
...     return "Foo"
```

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

### Run Remoulade Workers with Gevent

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/advanced.md

Utilize the remoulade-gevent utility to run workers with gevent for potential performance improvements, especially for I/O-bound tasks. Specify the number of worker processes and greenlets per process.

```bash
$ remoulade-gevent my_app -p 8 -t 250
```

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

### Broker.join

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Waits for all messages on a given queue to be processed. This method is intended for use in tests.

```APIDOC
## Broker.join(queue_name: str, , timeout: int | None = None) → None

### Description
Wait for all the messages on the given queue to be processed. This method is only meant to be used in tests to wait for all the messages in a queue to be processed.
```

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

### Set Actor Priority

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Define actor priorities using the `priority` keyword argument during actor registration. Higher numeric values indicate higher priority.

```python
@remoulade.actor(priority=1)
def generate_report(user_id):
  ...

@remoulade.actor(priority=0) # 0 is the default
def sync_order_to_warehouse(order_id):
  ...
```

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

### Set Actor Time Limit

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Define a maximum execution time for an actor in milliseconds. Actors exceeding this limit will be terminated with a TimeLimitExceeded error.

```python
@remoulade.actor(time_limit=60000)
def count_words(url):
  ...
```

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

### remoulade.Message.send_with_options

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Asynchronously send a message to this actor, along with an arbitrary set of processing options for the broker and middleware.

```APIDOC
## send_with_options

### Description
Asynchronously send a message to this actor, along with an arbitrary set of processing options for the broker and middleware.

### Parameters
#### Parameters
- **args** (tuple) - Positional arguments that are passed to the actor.
- **kwargs** (dict) - Keyword arguments that are passed to the actor.
- **queue_name** (str) - Name of the queue to enqueue this message into.
- **delay** (int) - The minimum amount of time, in milliseconds, the message should be delayed by.
- **\*\*options** (dict) - Arbitrary options that are passed to the broker and any registered middleware.

### Returns
- The enqueued message.
- **Return type:** [Message](#remoulade.Message)
```

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

### StubBroker Class

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

The StubBroker class provides a mock implementation of a message broker suitable for unit tests. It allows for the declaration of queues and actors, enqueuing and consuming messages, and managing middleware.

```APIDOC
## class remoulade.brokers.stub.StubBroker(middleware=None)

A broker that can be used within unit tests.

### Attributes
#### dead_letters
Contains the dead-lettered messages for all defined queues.
* **Type:** list[[Message](#remoulade.Message)]

### Methods
#### add_middleware(middleware: [Middleware](#remoulade.Middleware)) -> None
Add a middleware object to this broker. The middleware is added to its default position.
* **Parameters:**
  **middleware** ([*Middleware*](#remoulade.Middleware)) – The middleware.

#### close()
Close this broker and perform any necessary cleanup actions.

#### consume(queue_name, prefetch=1, timeout=100) -> [Consumer](#remoulade.Consumer)
Create a new consumer for a queue.
* **Parameters:**
  * **queue_name** (*str*) – The queue to consume.
  * **prefetch** (*int*) – The number of messages to prefetch.
  * **timeout** (*int*) – The idle timeout in milliseconds.
* **Raises:**
  [**QueueNotFound**](#remoulade.QueueNotFound) – If the queue hasn’t been declared.
* **Returns:**
  A consumer that retrieves messages from Redis.
* **Return type:**
  [Consumer](#remoulade.Consumer)

#### declare_actor(actor: [Actor](#remoulade.Actor)) -> None
Declare a new actor on this broker. Declaring an Actor twice replaces the first actor with the second by name.
* **Parameters:**
  **actor** ([*Actor*](#remoulade.Actor)) – The actor being declared.

#### declare_queue(queue_name)
Declare a queue. Has no effect if a queue with the given name has already been declared.
* **Parameters:**
  **queue_name** (*str*) – The name of the new queue.

#### enqueue(message: [Message](#remoulade.Message)[Any], delay: int | None = None) -> [Message](#remoulade.Message)[Any]
Enqueue a message on this broker.
* **Parameters:**
  * **message** ([*Message*](#remoulade.Message)) – The message to enqueue.
  * **delay** (*int*) – The number of milliseconds to delay the message for.
* **Returns:**
  Either the original message or a copy of it.
* **Return type:**
  [Message](#remoulade.Message)

#### flush(queue_name)
Drop all the messages from a queue.
* **Parameters:**
  **queue_name** (*str*) – The queue to flush.

#### flush_all()
Drop all messages from all declared queues.

#### get_actor(actor_name: str) -> [Actor](#remoulade.Actor)
Look up an actor by its name.
* **Parameters:**
  **actor_name** (*str*) – The name to look up.
* **Raises:**
  [**ActorNotFound**](#remoulade.ActorNotFound) – If the actor was never declared.
* **Returns:**
  The actor.
* **Return type:**
  [Actor](#remoulade.Actor)

#### get_cancel_backend() -> CancelBackend
Get the CancelBackend associated with the broker
* **Raises:**
  **NoCancelBackend** – if there is no CancelBackend
* **Returns:**
  the cancel backend
* **Return type:**
  CancelBackend

#### get_declared_actors() -> Set[str]
Get all declared actors.
* **Returns:**
  The names of all the actors declared so far on this Broker.
* **Return type:**
  set[str]

#### get_declared_delay_queues() -> Set[str]
Get all declared delay queues.
* **Returns:**
  The names of all the delay queues declared so far on this Broker.
* **Return type:**
  set[str]

#### get_declared_queues() -> Set[str]
Get all declared queues.
* **Returns:**
  The names of all the queues declared so far on this Broker.
* **Return type:**
  set[str]

#### get_result_backend() -> [ResultBackend](#remoulade.results.ResultBackend)
Get the ResultBackend associated with the broker
* **Raises:**
  [**NoResultBackend**](#remoulade.NoResultBackend) – if there is no ResultBackend
* **Returns:**
  the result backend
* **Return type:**
  [ResultBackend](#remoulade.results.ResultBackend)

#### get_state_backend() -> StateBackend
Get the StateBackend associated with the broker
* **Raises:**
  **NoStateBackend** – if there is no StateBackend
* **Returns:**
  the state backend
* **Return type:**
  StateBackend

#### join(queue_name, timeout=None)
Wait for all the messages on the given queue to be processed. This method is only meant to be used in tests to wait for all the messages in a queue to be processed.
* **Raises:**
  * **QueueJoinTimeout** – When the timeout elapses.
  * [**QueueNotFound**](#remoulade.QueueNotFound) – If the given queue was never declared.
* **Parameters:**
  * **queue_name** (*str*) – The queue to wait on.
  * **timeout** (*Optional* [**int**]) – The max amount of time, in milliseconds, to wait on this queue.

#### remove_middleware(middleware_class: Type[[Middleware](#remoulade.Middleware)])
Removes a middleware object from this broker.
* **Parameters:**
  **middleware_class** (*Type* [*[*Middleware*](#remoulade.Middleware) *]*) – The middleware class.
```

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

### Broker.add_middleware

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Adds a middleware object to the broker. The middleware is added to its default position.

```APIDOC
## Broker.add_middleware(middleware: [Middleware](#remoulade.Middleware)) → None

### Description
Add a middleware object to this broker. The middleware is added to its default position.

### Parameters:
- **middleware** ([*Middleware*](#remoulade.Middleware)) – The middleware.
```

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

### Schedule Periodic Messages with Remoulade

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/cookbook.md

Configure Remoulade to schedule recurring messages using a RabbitMQ broker and a Scheduler. Ensure the broker and scheduler are properly initialized and actors are declared.

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

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

### Generic Class Typing in Remoulade

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/changelog.md

Illustrates how various classes in Remoulade have become generic, including Message, Result, CollectionResult, pipeline, and group, reflecting their updated typing.

```python
from remoulade import actor, message, pipeline, group
from remoulade.result import Result, CollectionResult

@actor
def add(a: int, b: int) -> int:
   return a + b

add.message(1, 2)  # This is now Message[Result[int]]
add.message(1, 2).result  # this is now Result[int]
pipeline(message(1, 2), message(1))  # this is now pipeline[Result[int]]
group(message(1, 2), message(1))  # this is now group[Result[int]]
```

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

### Send Message to Alternative Queue

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/guide.md

Send a message to one of the alternative queues defined for an actor using the `send_with_options` method and specifying `queue_name`.

```python
count_words.send_with_options(queue_name="second_queue")
```

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

### remoulade.ConnectionError

Source: https://github.com/wiremind/remoulade/blob/master/docs/source/reference.md

Base class for errors related to establishing or maintaining connections with the broker. Provides a common base for connection-related exceptions.

```APIDOC
## class remoulade.ConnectionError(message: str)

### Description
Base class for broker connection-related errors.
```
Tokens:
Raw