dramatiq Documentation

pip install 'dramatiq[rabbitmq, watch]'

pip install 'dramatiq[redis, watch]'

import dramatiq
import requests
import sys


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


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

dramatiq example

python example.py http://example.com
python example.py https://github.com
python example.py https://news.ycombinator.com

RabbitMQ has the concept of `virtual hosts`_ built into it.  They
provide logical grouping and separation of resources.  You can create
virtual hosts using the `rabbitmqctl` command::

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

You can then pass that vhost to |RabbitmqBroker| when you instantiate it.

.. _virtual hosts:  https://www.rabbitmq.com/vhosts.html

Redis
~~~~~

The |RedisBroker| takes a ``namespace`` parameter that you can use to
logically split queues across multiple apps.


Highly Available Queues
^^^^^^^^^^^^^^^^^^^^^^^

RabbitMQ

Assuming you want to enqueue a message on a queue named ``default``,
publish a persistent message to that queue in RabbitMQ.

Using Redis
~~~~~~~~~~~

Add a field to the ``options`` dictionary of the message called
``redis_message_id`` that is a different UUID4 value than
``message_id``:

.. code-block:: javascript

   {
     ...
     "options": { "redis_message_id": "unique-id-2"},
     ...
   }

Assuming you want to enqueue a message on a queue named ``default``,
run::

  > HSET dramatiq:default.msgs $YOUR_REDIS_MESSAGE_ID $YOUR_MESSAGE_PAYLOAD
  > RPUSH dramatiq:default $YOUR_REDIS_MESSAGE_ID

``$YOUR_REDIS_MESSAGE_ID`` is the ``redis_message_id`` in the ``options``
field of the message payload.

Workers
-------

Worker Exit Codes
^^^^^^^^^^^^^^^^^

Dramatiq uses process exit codes to denote several scenarios:

=====  ========================================================================================
Code   Description
=====  ========================================================================================
``0``  Returned when the process exits gracefully.
``1``  Returned when the process is killed.
``2``  Returned when a module cannot be imported or when a command line argument is invalid.
``3``  Returned when a broker connection cannot be established during worker startup.
``4``  Returned when a PID file is set and Dramatiq is already running.
=====  ========================================================================================


Controlling Workers
^^^^^^^^^^^^^^^^^^^

The main Dramatiq process responds to several signals::

  $ kill -TERM [master-process-pid]

``INT`` and ``TERM``

Sending ``HUP`` to the main process triggers a graceful shutdown
followed by a reload of the workers.  This is useful if you want to
reload code without completely restarting the main process.


Using gevent
^^^^^^^^^^^^

Dramatiq comes with a CLI utility called ``dramatiq-gevent`` that can
run workers under gevent_.  The following invocation would run 8
worker processes with 250 greenlets per process for a total of 2k
lightweight worker threads::

  $ dramatiq-gevent my_app -p 8 -t 250

If your tasks spend most of their time doing network IO and don't
depend on C extensions to execute those network calls then using
gevent could provide a significant performance improvement.

I suggest at least experimenting with it to see if it fits your use
case.

.. _prometheus-metrics:

Prometheus Metrics
^^^^^^^^^^^^^^^^^^

Prometheus metrics are automatically exported by workers whenever you
run them using the command line utility (assuming you're using the
Prometheus middleware).  By default, the exposition server listens on
port ``9191`` so you can tell Prometheus to scrape that or you can
specify what host and port it should listen on by setting the
``dramatiq_prom_host`` and ``dramatiq_prom_port`` environment
variables.

The following metrics are exported:

``dramatiq_messages_total``
  A *counter* for the total number of messages processed.

``dramatiq_message_errors_total``
  A *counter* for the total number of errored messages.

``dramatiq_message_retries_total``
  A *counter* for the total number of retried messages.

``dramatiq_message_rejects_total``
  A *counter* for the total number of dead-lettered messages.

``dramatiq_messages_inprogress``
  A *gauge* for the number of messages currently being processed.

``dramatiq_delayed_messages_inprogress``
  A *gauge* for the number of delayed messages currently in memory.

``dramatiq_message_duration_milliseconds``
  A *histogram* for the time spent processing messages.

All metrics define labels for ``queue_name`` and ``actor_name``.

Grafana Dashboard

The Prometheus client for Python is a bit finicky when it comes to
exporting metrics from a multi-process configuration.  If your own app
uses Prometheus, then you should export ``prometheus_multiproc_dir``
and ``dramatiq_prom_db`` environment variables -- both pointing to an
existing folder -- and remove any files in that folder before running
Dramatiq.  For example::

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

If you don't do this, then metrics will likely fail to export properly.

Workflows
---------

Nested Pipelines and Groups
^^^^^^^^^^^^^^^^^^^^^^^^^^^

Dramatiq supports running tasks in parallel via |group| and in
sequence via |pipeline|, but it does not provide a built-in way to run
a pipeline of groups out of the box. If you need more advanced
orchestration capabilities, such as nesting chains and groups of tasks
at arbitrary depth, you may find the dramatiq-workflow_ package
useful.

.. _dramatiq-workflow: https://github.com/Outset-AI/dramatiq-workflow

---

## File: docs/source/best_practices.rst

.. include:: global.rst

Best Practices
==============

Concurrent Actors
-----------------

Your actor will run concurrently with other actors in the system.  You
need to be mindful of the impact this has on your database, any third
party services you might be calling and the resources available on the
systems running your workers.  Additionally, you need to be mindful of
data races between actors that manipulate the same objects in your
database.


Retriable Actors
----------------

Dramatiq actors may receive the same message multiple times in the
event of a worker failure (hardware, network or power failure).  This
means that, for any given message, running your actor multiple times
must be safe.  This is also known as being "idempotent".


Simple Messages
---------------

Attempting to send an actor any object that can't be encoded to JSON
by the standard ``json`` package will fail immediately so you'll want
to limit your actor parameters to the following object types: `bool`,
`int`, `float`, `bytes`, `string`, `list` and `dict`.

Additionally, since messages are sent over the wire you'll want to
keep them as short as possible.  For example, if you've got an actor
that operates over ``User`` objects in your system, send that actor
the user's id rather than the serialized user.
Or, if you have an actor that operates on a file, send the path to the
file in the message, rather than the body of the file.


Error Reporting
---------------

Invariably, you're probably going to introduce issues in production
every now and then and some of those issues are going to affect your
tasks.  You should use an error reporting service such as Sentry_ so
you get notified of these errors as soon as they occur.


.. _Sentry: https://sentry.io/welcome/

---

## File: docs/source/cookbook.rst

.. include:: global.rst

Cookbook
========

This part of the docs contains recipes for various things you might
want to do using Dramatiq.  Each section will be light on prose and
code heavy, so if you have any questions about one of the recipes,
open an `issue on GitHub`_.

.. _issue on GitHub: https://github.com/Bogdanp/dramatiq/issues


Callbacks
---------

Dramatiq has built-in support for sending actors messages when other
actors succeed or fail.  The ``on_failure`` callback is called every
time an actor fails, even if the message is going to be retried.

.. code-block:: python

   import dramatiq


   @dramatiq.actor
   def identity(x):
       return x


   @dramatiq.actor
   def print_result(message_data, result):
       print(f"The result of message {message_data['message_id']} was {result}.")

   @dramatiq.actor
   def print_error(message_data, exception_data):
       print(f"Message {message_data['message_id']} failed:")
       print(f"  * type: {exception_data['type']}")
       print(f"  * message: {exception_data['message']!r}")


   if __name__ == "__main__":
       identity.send_with_options(
           args=(42,),
           on_failure=print_error,
           on_success=print_result,
       )


Composition
-----------

Dramatiq has built-in support for a couple high-level composition
constructs.  You can use these to combine generalized tasks that don't
know about one another into complex workflows.

In order to take advantage of group and pipeline result management
or to wait for a group or pipeline to finish, you need to enable
result storage and your actors need to store results.  Check out the
`Results`_ section for more information on result storage.

Groups
^^^^^^

|Groups| run actors in parallel and let you gather their results or
wait for all of them to finish.  Assuming you have a computationally
intensive actor called ``frobnicate``, you can group multiple
messages together as follows::

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

This will enqueue 3 separate messages and, assuming there are enough
resources available, execute them in parallel.  You can then wait for
the whole group to finish::

  g.wait(timeout=10_000)  # 10s expressed in millis

Or you can iterate over the results::

  for res in g.get_results(block=True, timeout=5_000):
      ...

Results are returned in the same order that the messages were added to
the group.

Pipelines
^^^^^^^^^

Actors can be chained together using the |pipeline| function.  For
example, if you have an actor that gets the text contents of a website
and one that counts the number of "words" in a piece of text:

.. code-block:: python

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


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

You can chain them together like so::

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

Or you can use pipe notation to achieve the same thing::

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

In both cases, the result of running ``get_uri_contents(uri)`` is
passed as the last positional argument to ``count_words``.  If you
would like to avoid passing the result of an actor to the next one in
line, set the ``pipe_ignore`` option to ``True`` when you create the
"receiving" message::

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

Here, the result of ``bust_caches()`` will not be passed to
``prepare_codes()``, but the result of ``prepare_codes()`` will be
passed to ``launch_missiles(codes)``.  To get the end result of a
pipeline -- that is, the result of the last actor in the pipeline --
you can call |pipeline_get_result|::

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

To get the intermediate results of each step in the pipeline, you can
call |pipeline_get_results|::

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

Asyncio Actors
--------------

To use asynchronous functions (``async def``) as actors,
you must enable the optional |AsyncIO| middleware (See :ref:`customizing-middleware`).

.. code-block:: python

   import asyncio
   import dramatiq
   import dramatiq.middleware

   broker = dramatiq.Broker()
   dramatiq.set_broker(broker)
   broker.add_middleware(dramatiq.middleware.AsyncIO())

   @dramatiq.actor
   async def waiter():
       await asyncio.sleep(10)

This middleware will start up 1 thread per worker process to run an event loop,
where any asynchronous actor functions will be scheduled to run.

You can use a mixture of synchronous actors and asynchronous actors,
the synchronous ones will be run on the worker threads as normal,
and the asynchronous ones will be scheduled on the event loop thread.

Note that the maximum number of concurrent messages processed by a worker process,
is still equal to the number of worker threads (8 threads by default).
This is because each worker thread waits for the result of an asynchronous actor
(same as a synchronous actor) before starting the next message.

Error Reporting
---------------

Reporting errors with Rollbar
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Rollbar_ provides an easy-to-use Python client::

   $ pip install rollbar

Save the following middleware to a module inside your project:

.. code-block:: python

   import dramatiq
   import rollbar


   class RollbarMiddleware(dramatiq.Middleware):
       def after_process_message(self, broker, message, *, result=None, exception=None):
           if exception is not None:
               rollbar.report_exc_info()

Finally, instantiate and add it to your broker:

.. code-block:: python

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


.. _Rollbar: https://github.com/rollbar/pyrollbar#quick-start

Reporting errors with Sentry
^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Use `sentry-sdk`_ and follow the `Dramatiq integration docs`_.

.. _sentry-sdk: https://pypi.org/project/sentry-sdk/
.. _Dramatiq integration docs: https://docs.sentry.io/platforms/python/integrations/dramatiq/


Frameworks
----------

API Star
^^^^^^^^

The `apistar_dramatiq`_ library lets you use API Star dependency
injection with your Dramatiq actors.

.. _apistar_dramatiq: https://github.com/Bogdanp/apistar_dramatiq
.. _API Star: https://github.com/encode/apistar


Django
^^^^^^

Check out the `django_dramatiq`_ project if you want to use Dramatiq
with Django_.  The `django_dramatiq_example`_ repo is an example app
build with Django and Dramatiq.

.. _django_dramatiq: https://github.com/Bogdanp/django_dramatiq
.. _django_dramatiq_example: https://github.com/Bogdanp/django_dramatiq_example
.. _django: https://djangoproject.com


Flask
^^^^^

The `Flask-Dramatiq`_ extension integrates Dramatiq with Flask_.  It
includes support for configuration and the application factory
pattern.  `Flask-Melodramatiq`_ is very similar in scope, but tries to
stay as close as possible to the "native" Dramatiq API.

.. _Flask-Dramatiq: https://flask-dramatiq.readthedocs.io
.. _Flask-Melodramatiq: https://flask-melodramatiq.readthedocs.io
.. _flask: http://flask.pocoo.org


Operations
----------

Auto-discovering "tasks" modules with bash
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Dramatiq doesn't attempt to auto-discover tasks modules.  Assuming you
follow a convention where all your tasks modules are named
``tasks.py`` then you can discover them using bash:

.. code-block:: bash

   #!/usr/bin/env bash

   set -e

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

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

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

Retrying connection errors on startup
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Dramatiq does not retry connection errors that occur on worker
startup.  It does, however, return a specific exit code (``3``) when
that happens.  Using that, you can build a wrapper script around it if
you need to retry with backoff when connection errors happen during
startup (eg. in Docker):

.. code-block:: bash

   #!/usr/bin/env bash

   delay=1
   while true; do
     dramatiq $@
     if [ $? -eq 3 ]; then
       echo "Connection error encountered on startup. Retrying in $delay second(s)..."
       sleep $delay
       delay=$((delay * 2))
     else
       exit $?
     fi
   done

Binding Worker Groups to Queues
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

By default, Dramatiq workers consume all declared queues, but it's
common to want to bind worker groups to specific queues in order to
have better control over throughput.  For example, given the following
actors::

   @dramatiq.actor
   def very_slow():
       ...


   @dramatiq.actor(queue_name="ui-blocking")
   def very_important():
       ...

You may want to run one group of workers that only processes messages
on the ``default`` queue and another that only processes messages off
of the ``ui-blocking`` queue.  To do that, you have to pass each group
the appropriate queue on the command line:

.. code-block:: bash

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

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

Messages sent to ``very_slow`` will always be delivered to those
workers that consume the ``default`` queue and messages sent to
``very_important`` will always be delivered to the ones that consume
the ``ui-blocking`` queue.

Rate Limiting
-------------

Rate limiting work
^^^^^^^^^^^^^^^^^^

You can use Dramatiq's |RateLimiters| to constrain actor concurrency.

.. code-block:: python

   import dramatiq
   import time

   from dramatiq.rate_limits import ConcurrentRateLimiter
   from dramatiq.rate_limits.backends import RedisBackend

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


   @dramatiq.actor
   def one_at_a_time():
       with DISTRIBUTED_MUTEX.acquire():
           time.sleep(1)
           print("Done.")

Whenever two ``one_at_a_time`` actors run at the same time, one of
them will be retried with exponential backoff.  This works by raising
an exception and relying on the built-in
:class:`Retries<dramatiq.middleware.Retries>` middleware to do the
work of re-enqueueing the task.

If you want rate limiters not to raise an exception when they can't be
acquired, you should pass ``raise_on_failure=False`` to ``acquire``::

  with DISTRIBUTED_MUTEX.acquire(raise_on_failure=False) as acquired:
      if not acquired:
          print("Lock could not be acquired.")
      else:
          print("Lock was acquired.")


Results
-------

Storing message results
^^^^^^^^^^^^^^^^^^^^^^^

You can use Dramatiq's result backends to store and retrieve message
return values.  To enable result storage, you need to instantiate and
add the |Results| middleware to your broker.

.. code-block:: python

   import dramatiq

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

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


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


   if __name__ == "__main__":
       message = add.send(1, 2)
       print(message.get_result(block=True))

Getting a result raises |ResultMissing| when a result hasn't been
stored yet or if it has already expired (results expire after 10
minutes by default).  When the ``block`` parameter is ``True``,
|ResultTimeout| is raised instead.

When a message fails, getting a result raises |ResultFailure|.  When a
message is skipped via |SkipMessage|, ``None`` is stored as the
result.

Results expire, otherwise the result backend would eventually run out
of space.  The timeout for the results expiration is set on
`result_ttl` argument of |Results|, given in milliseconds, and the
default is 10 minutes.

.. code-block:: python

   # Results are valid only for one hour
   Results(backend=result_backend, result_ttl=3600*1000)

The result expiration can be also set per an actor:

.. code-block:: python

   # Add result expires in 30 seconds
   @dramatiq.actor(store_results=True, result_ttl=30*1000)
   def add(x, y):
       return x + y


.. _result-key-format:

Result Key Format
^^^^^^^^^^^^^^^^^

The ``use_namespace_prefix_keys`` option of the |ResultBackends| controls
the format of the keys used to store results in the backed.
Specifically, it toggles between keys using a structured format,
or keys being plain MD5 hashes.

.. caution::

   Dramatiq does not do any migration of result data
   if you change the value of ``use_namespace_prefix_keys``.
   Results stored when ``use_namespace_prefix_keys=False`` won't
   be read when ``use_namespace_prefix_keys=True``, and vice versa.

If you need complete control over the format of result keys, you can subclass
the result backend class you are using and override
:meth:`ResultBackend.build_message_key<dramatiq.results.ResultBackend.build_message_key>`:

.. code-block:: python

   import dramatiq

   from dramatiq.results.backends import RedisBackend
   from dramatiq.results import Results

   class CustomRedisBackend(RedisBackend):
       def build_message_key(self, message):
           return f"my_custom_key_{message.message_id}"

   result_backend = CustomRedisBackend()
   broker.add_middleware(Results(backend=result_backend))


Custom Middleware
-----------------

Writing a Custom Middleware Class
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

You can write a custom middleware class to customize or add functionality to many parts of Dramatiq.
For the full list of hooks available, see the |Middleware| class.

For example, you could write a middleware that sets the message time limit dynamically based
on the message arguments,
using the :meth:`Middleware.before_process_message<dramatiq.Middleware.before_process_message>` hook.

.. code-block:: python

   import dramatiq

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

Finally, instantiate and add it to your broker:

.. code-block:: python

   import dramatiq.middleware

   broker.add_middleware(
       DynamicTimeLimitMiddleware(),
       before=dramatiq.middleware.TimeLimit,
   )

Note that in this case, the custom middleware must be inserted before the |TimeLimit| middleware
so the dynamic time limit is applied to messaged before it is used by the |TimeLimit| middleware.


Scheduling
----------

Scheduling messages
^^^^^^^^^^^^^^^^^^^

APScheduler_ is the recommended scheduler to use with Dramatiq:

.. code-block:: python

   import dramatiq
   import sys

   from apscheduler.schedulers.blocking import BlockingScheduler
   from apscheduler.triggers.cron import CronTrigger
   from datetime import datetime


   @dramatiq.actor
   def print_current_date():
       print(datetime.now())


   if __name__ == "__main__":
       scheduler = BlockingScheduler()
       scheduler.add_job(
           print_current_date.send,
           CronTrigger.from_crontab("* * * * *"),
       )
       try:
           scheduler.start()
       except KeyboardInterrupt:
           scheduler.shutdown()

.. _APScheduler: https://apscheduler.readthedocs.io


Aborting
--------

Aborting Messages
^^^^^^^^^^^^^^^^^

The `dramatiq-abort`_ package provides a middleware that can be used
to abort running actors by message id.  Here's how you might set it
up::

  import dramatiq
  import dramatiq_abort.backends
  from dramatiq_abort import Abortable, abort

  abortable = Abortable(
      backend=dramatiq_abort.backends.RedisBackend()
  )
  dramatiq.get_broker().add_middleware(abortable)

  @dramatiq.actor
  def a_long_running_task():
      ...

  message = a_long_running_task.send()
  abort(message.message_id)

.. _`dramatiq-abort`: https://flared.github.io/dramatiq-abort/

---

## File: docs/source/global.rst

.. References

.. |AgeLimit| replace:: :class:`AgeLimit<dramatiq.middleware.AgeLimit>`
.. |AsyncIO| replace:: :class:`AsyncIO<dramatiq.middleware.AsyncIO>`
.. |Barriers| replace:: :class:`Barriers<dramatiq.rate_limits.Barrier>`
.. |Brokers| replace:: :class:`Brokers<dramatiq.Broker>`
.. |Broker| replace:: :class:`Broker<dramatiq.Broker>`
.. |Callbacks| replace:: :class:`Callbacks<dramatiq.middleware.Callbacks>`
.. |CurrentMessage| replace:: :class:`CurrentMessage<dramatiq.middleware.CurrentMessage>`
.. |DramatiqError| replace:: :class:`DramatiqError<dramatiq.DramatiqError>`
.. |Encoders| replace:: :class:`Encoders<dramatiq.Encoder>`
.. |GenericActors| replace:: :class:`class-based actors<dramatiq.GenericActor>`
.. |Groups| replace:: :func:`Groups<dramatiq.group>`
.. |GroupCallbacks| replace:: :class:`GroupCallbacks<dramatiq.middleware.GroupCallbacks>`
.. |Interrupt| replace:: :class:`Interrupt<dramatiq.middleware.Interrupt>`
.. |MemcachedRLBackend| replace:: :class:`Memcached<dramatiq.rate_limits.backends.MemcachedBackend>`
.. |Messages| replace:: :class:`Messages<dramatiq.Message>`
.. |Middleware| replace:: :class:`Middleware<dramatiq.Middleware>`
.. |MiddlewareError| replace:: :class:`MiddlewareError<dramatiq.middleware.MiddlewareError>`
.. |Pipelines| replace:: :class:`Pipelines<dramatiq.middleware.Pipelines>`
.. |Prometheus| replace:: :class:`Prometheus<dramatiq.middleware.prometheus.Prometheus>`
.. |RabbitmqBroker_join| replace:: :meth:`join<dramatiq.brokers.rabbitmq.RabbitmqBroker.join>`
.. |RabbitmqBroker| replace:: :class:`RabbitmqBroker<dramatiq.brokers.rabbitmq.RabbitmqBroker>`
.. |RateLimitExceeded| replace:: :class:`RateLimitExceeded<dramatiq.RateLimitExceeded>`
.. |RateLimiters| replace:: :class:`RateLimiters<dramatiq.rate_limits.RateLimiter>`
.. |RedisBroker| replace:: :class:`RedisBroker<dramatiq.brokers.redis.RedisBroker>`
.. |RedisRLBackend| replace:: :class:`Redis<dramatiq.rate_limits.backends.RedisBackend>`
.. |RedisResBackend| replace:: :class:`Redis<dramatiq.results.backends.RedisBackend>`
.. |ResultBackends| replace:: :class:`ResultBackends<dramatiq.results.ResultBackend>`
.. |ResultBackend| replace:: :class:`ResultBackend<dramatiq.results.ResultBackend>`
.. |ResultFailure| replace:: :class:`ResultFailure<dramatiq.results.ResultFailure>`
.. |ResultMissing| replace:: :class:`ResultMissing<dramatiq.results.ResultMissing>`
.. |ResultTimeout| replace:: :class:`ResultTimeout<dramatiq.results.ResultTimeout>`
.. |Results| replace:: :class:`Results<dramatiq.results.Results>`
.. |Retries| replace:: :class:`Retries<dramatiq.middleware.Retries>`
.. |ShutdownNotifications| replace:: :class:`ShutdownNotifications<dramatiq.middleware.ShutdownNotifications>`
.. |Shutdown| replace:: :class:`Shutdown<dramatiq.middleware.Shutdown>`
.. |SkipMessage| replace:: :class:`SkipMessage<dramatiq.middleware.SkipMessage>`
.. |StubBroker_flush_all| replace:: :meth:`StubBroker.flush_all<dramatiq.brokers.stub.StubBroker.flush_all>`
.. |StubBroker_flush| replace:: :meth:`StubBroker.flush<dramatiq.brokers.stub.StubBroker.flush>`
.. |StubBroker_join| replace:: :meth:`StubBroker.join<dramatiq.brokers.stub.StubBroker.join>`
.. |StubBroker| replace:: :class:`StubBroker<dramatiq.brokers.stub.StubBroker>`
.. |TimeLimitExceeded| replace:: :class:`TimeLimitExceeded<dramatiq.middleware.TimeLimitExceeded>`
.. |TimeLimit| replace:: :class:`TimeLimit<dramatiq.middleware.TimeLimit>`
.. |WindowRateLimiter| replace:: :class:`WindowRateLimiter<dramatiq.rate_limits.WindowRateLimiter>`
.. |Worker_join| replace:: :meth:`Worker.join<dramatiq.Worker.join>`
.. |Worker_pause| replace:: :meth:`Worker.pause<dramatiq.Worker.pause>`
.. |Worker_resume| replace:: :meth:`Worker.resume<dramatiq.Worker.resume>`
.. |Worker| replace:: :meth:`Worker<dramatiq.Worker>`
.. |actor| replace:: :func:`actor<dramatiq.actor>`
.. |add_middleware| replace:: :meth:`add_middleware<dramatiq.Broker.add_middleware>`
.. |after_skip_message| replace:: :meth:`after_skip_message<dramatiq.Middleware.after_skip_message>`
.. |before_consumer_thread_shutdown| replace:: :meth:`before_consumer_thread_shutdown<dramatiq.Middleware.before_consumer_thread_shutdown>`
.. |before_worker_thread_shutdown| replace:: :meth:`before_worker_thread_shutdown<dramatiq.Middleware.before_worker_thread_shutdown>`
.. |dramatiq| replace:: :mod:`dramatiq`
.. |get_current_message| replace:: :meth:`get_current_message<dramatiq.middleware.CurrentMessage.get_current_message>`
.. |group| replace:: :func:`group<dramatiq.group>`
.. |pipeline_get_results| replace:: :meth:`get_results<dramatiq.pipeline.get_results>`
.. |pipeline_get_result| replace:: :meth:`get_result<dramatiq.pipeline.get_result>`
.. |pipeline| replace:: :func:`pipeline<dramatiq.pipeline>`
.. |rate_limits| replace:: :mod:`dramatiq.rate_limits`
.. |send_with_options| replace:: :meth:`send_with_options<dramatiq.Actor.send_with_options>`
.. |send| replace:: :meth:`send<dramatiq.Actor.send>`

.. _gevent: http://www.gevent.org/
.. _Memcached: http://memcached.org
.. _RabbitMQ: https://www.rabbitmq.com
.. _Redis: https://redis.io

---

## File: docs/source/guide.rst

.. include:: global.rst

User Guide
==========

To follow along with this guide you'll need to install and run RabbitMQ_
and then set up a new `virtual environment`_ in which you'll have
to install Dramatiq and Requests_::

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

.. _requests: http://docs.python-requests.org
.. _virtual environment: http://docs.python-guide.org/en/latest/starting/install3/osx/#virtual-environments


Actors
------

As a quick and dirty example of a task that's worth processing in the
background, let's write a function that counts the "words" at a
particular URL:

.. code-block:: python
   :caption: count_words.py

   import requests


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

There's not a ton going on here.  We just grab the response content at
that URL and print out how many space-separated chunks there are in
that response.  Sure enough, running this in the interactive
interpreter yields about what we expect::

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

To turn this into a function that can be processed asynchronously
using Dramatiq, all we have to do is decorate it with |actor|:

.. code-block:: python
   :caption: count_words.py
   :emphasize-lines: 1, 5

   import dramatiq
   import requests


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

Like before, if we call the function in the interactive interpreter,
it will run synchronously and we'll get the same result out::

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

What's changed is we're now able to tell the function to run
asynchronously by calling its |send| method::

  >>> count_words.send("http://example.com")
  Message(
    queue_name='default',
    actor_name='count_words',
    args=('http://example.com',), kwargs={}, options={},
    message_id='8cdcae57-af36-40ba-9616-849a336a4316',
    message_timestamp=1498557015410)

Doing so immediately enqueues a message (via our local RabbitMQ
server) that can be processed asynchronously but *doesn't actually run
the function*.  In order to run it, we'll have to boot up a Dramatiq
worker.

.. note::
   Because all messages have to be sent over the network, any
   arguments you send to an actor must be JSON-encodable.


Workers
-------

Dramatiq comes with a command line utility called, predictably,
``dramatiq``.  This utility is able to spin up multiple concurrent
worker processes that pop messages off the queue and send them to
actor functions for execution.

To spawn workers for our ``count_words.py`` example, run the following
command in a new terminal window::

  $ dramatiq count_words

This will spin up as many processes as there are CPU cores on your
machine with 8 worker threads per process.  Run ``dramatiq -h`` if you
want to see a list of the available command line flags.

As soon as you run that command you'll see log output along these
lines::

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

If you open your Python interpreter back up and send the actor some
more URLs to process::

  >>> urls = [
  ...     "https://news.ycombinator.com",
  ...     "https://xkcd.com",
  ...     "https://rabbitmq.com",
  ... ]
  >>> [count_words.send(url) for url in urls]
  [Message(queue_name='default', actor_name='count_words', args=('https://news.ycombinator.com',), kwargs={}, options={}, message_id='a99a5b2d-d2da-407b-be55-f2925266e216', message_timestamp=1498557998218),
   Message(queue_name='default', actor_name='count_words', args=('https://xkcd.com',), kwargs={}, options={}, message_id='0ec93dcb-2f9f-414f-99ec-7035e3b1ac5a', message_timestamp=1498557998218),
   Message(queue_name='default', actor_name='count_words', args=('https://rabbitmq.com',), kwargs={}, options={}, message_id='d3dd9799-1ea5-4b00-a70b-2cd6f6f634ed', message_timestamp=1498557998218)]

and then switch back to the worker terminal, you'll see nine new
lines::

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

At this point, you're probably wondering what happens if you send the
actor an invalid URL.  Let's try it::

  >>> count_words.send("foo")


Error Handling
--------------

Dramatiq assumes all actors are idempotent so when an exception occurs
during message processing, it automatically enqueues a retry for that
message with exponential backoff.

That last message we sent will cause something along these lines to be
printed in your worker process::

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

Dramatiq will keep retrying the message with longer and longer delays
in between runs until we fix our code or for up to about 30 days from
when it was first enqueued.

Change ``count_words`` to catch the missing schema error::

   @dramatiq.actor
   def count_words(url):
       try:
           response = requests.get(url)
           count = len(response.text.split(" "))
           print(f"There are {count} words at {url!r}.")
       except requests.exceptions.MissingSchema:
           print(f"Message dropped due to invalid url: {url!r}")

Then send ``SIGHUP`` to the main worker process to make the workers
pick up the source code changes::

  $ kill -s HUP 13047

Substitute the process ID of your own main process for ``13047``.  You
can find the PID by looking at the log lines from the worker starting
up.  Look for lines containing the string ``[dramatiq.MainProcess]``.

The next time your message is retried you should see::

  Message dropped due to invalid url: 'foo'


Code Reloading
--------------

Sending ``SIGHUP`` to the workers every time you make a change is
going to get old quick.  Instead, you can run the command line utility
with the ``--watch`` flag pointing to the folder it should watch for
source code changes.  It'll reload the workers whenever Python files
under that folder or any of its sub-folders change::

  $ dramatiq count_words --watch .

.. warning::
   Although this is a handy feature to use when developing your code,
   you should avoid using it in production!

.. _message-retries:

Message Retries
---------------

As mentioned in the error handling section, Dramatiq automatically
retries failures with exponential backoff.

You can specify how failures should be retried on a per-actor basis.
For example, if you want to limit the maximum number of retries for
``count_words`` you can pass the ``max_retries`` keyword argument to
|actor|::

  @dramatiq.actor(max_retries=3)
  def count_words(url):
    ...

If you want to retry certain exceptions and not others, you can pass a
predicate function via the ``retry_when`` parameter::

  def should_retry(retries_so_far, exception):
      return retries_so_far < 3 and isinstance(exception, HttpTimeout)


  @dramatiq.actor(retry_when=should_retry)
  def count_words(url):
      ...

The following retry options are configurable on a per-actor basis:

===============  ============  =====================================================================================================================
Option           Default       Description
===============  ============  =====================================================================================================================
``max_retries``  ``20``        The maximum number of times a message should be retried.  ``None`` means the message should be retried indefinitely.
``min_backoff``  15 seconds    The minimum number of milliseconds of backoff to apply between retries.  Must be greater than 100 milliseconds.
``max_backoff``  7 days        The maximum number of milliseconds of backoff to apply between retries.  Higher values are less reliable.
``retry_when``   ``None``      A callable that determines whether or not a message should be retried.  When this is set, ``max_retries`` is ignored.
``throws``       ``None``      An exception or a tuple of exceptions that must not get retried if they are raised from within the actor.
===============  ============  =====================================================================================================================


Message Age Limits
------------------

Instead of limiting the number of times messages can be retried, you
might want to expire old messages.  You can specify the ``max_age`` of
messages (given in milliseconds) on a per-actor basis::

  @dramatiq.actor(max_age=3600000)
  def count_words(url):
      ...

Dead Letters
^^^^^^^^^^^^

Once a message has exceeded its retry or age limits, it gets moved to
the dead letter queue where it's kept for up to 7 days and then
automatically dropped from the message broker.  From here, you can
manually inspect the message and decide whether or not it should be
put back on the queue.

.. _message-time-limits:

Message Time Limits
-------------------

In ``count_words``, we didn't set an explicit timeout for the outbound
request which means that it can take a very long time to complete if
the server we're requesting is timing out.  Dramatiq has a default
actor time limit of 10 minutes, which means that any actor running for
longer than 10 minutes is killed with a |TimeLimitExceeded| error.

You can control these time limits at the individual actor level by
specifying the ``time_limit`` (in milliseconds) of each one::

  @dramatiq.actor(time_limit=60000)
  def count_words(url):
      ...

.. note::
   While this will keep our actor from running forever, remember that
   you should take care to always specify a timeout for the request
   itself, and this is **not** a good way to handle request timeouts
   in production code.

.. warning::
   Time limits are best-effort.  They cannot cancel system calls or
   any function that doesn't currently hold the GIL under CPython.

   For more information, see the section on :ref:`message-interrupts`.


Handling Time Limits
^^^^^^^^^^^^^^^^^^^^

If you want to gracefully handle time limits within an actor, you can
wrap its source code in a try block and catch |TimeLimitExceeded|::

  from dramatiq.middleware import TimeLimitExceeded


  @dramatiq.actor(time_limit=1000)
  def long_running():
      try:
          setup_missiles()
          time.sleep(2)
          launch_missiles()    # <- this will not run
      except TimeLimitExceeded:
          teardown_missiles()  # <- this will run


Scheduling Messages
-------------------

You can schedule messages to run some time in the future by calling
|send_with_options| on actors and providing a ``delay`` (in
milliseconds)::

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

Keep in mind that *your message broker is not a database*.  Scheduled
messages should represent a small subset of all your messages.


Prioritizing Messages
---------------------

Say your app has some actors that are higher priority than others: for
example, actors that affect your UI and make users wait, or are
otherwise user-facing, versus actors that aren't.  When choosing
between multiple messages to run, Dramatiq will run the message
that belongs to the actor with the highest priority.

.. warning::
   Actor priority only takes effect when Dramatiq is choosing which message to run.
   This only happens *after* messages are consumed from the queues.
   The actory priority option has **no effect on messages while they are waiting in a queue**.
   It only applies once messaged are consumed from the queues, and are waiting in-memory to be processed.

You can set an Actor's priority via the ``priority`` keyword argument::

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


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

That way if messages for both ``generate_report`` and ``sync_order_to_warehouse``
have been consumed from the queue at the same time but there's only capacity to run
one of them, ``generate_report`` will always run *first*.

Although all positive integers represent valid priorities, if you're
going to use this feature, I'd recommend setting up constants for the
various priorities you plan to use::

  PRIORITY_LOW = 100
  PRIORITY_MEDIUM = 50
  PRIORITY_HIGH = 0

.. important::
   The lower the numeric value, the higher priority!  If you need a
   mnemonic to remember this, think of having a ticket with a number
   on it handed to you while waiting in line, perhaps at a government
   institution or deli counter.

Because actor ``priority`` does not effect messages waiting in the queues,
it is most useful for making Dramatiq treat certain queues as higher priority than others.
To make this work best, all the actors for a queue should have the same actor ``priority``.

.. seealso::
   The |RabbitmqBroker| supports prioritizing messages while they are waiting in the queue
   with the ``broker_priority`` option.


Message Brokers
---------------

Dramatiq abstracts over the notion of a message broker and currently
supports both RabbitMQ and Redis out of the box.  By default, it'll
set up a RabbitMQ broker instance pointing at the local host.

RabbitMQ Broker
^^^^^^^^^^^^^^^

To configure the RabbitMQ host, instantiate a |RabbitmqBroker| and set
it as the global broker as early as possible during your program's
execution::

  import dramatiq

  from dramatiq.brokers.rabbitmq import RabbitmqBroker


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

Redis Broker
^^^^^^^^^^^^

To use Dramatiq with the Redis broker, create an instance of it and
set it as the global broker as early as possible during your program’s
execution::

  import dramatiq

  from dramatiq.brokers.redis import RedisBroker


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


Unit Testing
------------

Dramatiq provides a |StubBroker| that can be used in unit tests so you
don't have to have a running RabbitMQ or Redis instance in order to
run your tests.  My recommendation is to use it in conjunction with
`pytest fixtures`_:

.. code-block:: python
   :caption: broker.py

   import os

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


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

.. code-block:: python
   :caption: conftest.py

   import dramatiq
   import pytest

   from dramatiq import Worker
   from yourapp import broker


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


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

Then you can inject and use those fixtures in your tests::

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

Because all actors are callable, you can of course also unit test them
synchronously by calling them as you would normal functions.

Dealing with Exceptions
^^^^^^^^^^^^^^^^^^^^^^^^

By default, any exceptions raised by an actor are caught by the
worker, which runs in a separate thread from the one your tests run
in.  This means that any exceptions your actor throws will not be
immediately visible to your test code!

To help surface actor exceptions, by default,
the stub broker will re-raise exceptions from failed messages
in your main thread when you call its |StubBroker_join| method::

  def test_count_words(stub_broker, stub_worker):
      count_words.send("http://some-invalid-url.invalid")
      stub_broker.join(count_words.queue_name)  # Exception from actor will be re-raised here.
      stub_worker.join()

This way, whatever exception caused the actor to fail will be raised
eagerly within your test.  Note that the exception will only be raised
once the actor exceeds its available retries.


.. _pytest fixtures: https://docs.pytest.org/en/latest/fixture.html

---

## File: docs/source/index.rst

.. include:: global.rst

Dramatiq: background tasks
==========================

Release v\ |release|. (:doc:`installation`, :doc:`changelog`, `Discuss`_, `Source Code`_)

.. _Discuss: https://groups.io/g/dramatiq-users
.. _Source Code: https://github.com/Bogdanp/dramatiq

.. image:: https://img.shields.io/badge/license-LGPL-blue.svg
   :target: license.html
.. image:: https://github.com/Bogdanp/dramatiq/workflows/CI/badge.svg
   :target: https://github.com/Bogdanp/dramatiq/actions?query=workflow%3A%22CI%22
.. image:: https://badge.fury.io/py/dramatiq.svg
   :target: https://badge.fury.io/py/dramatiq

**Dramatiq** is a background task processing library for Python with a
focus on simplicity, reliability and performance.

.. raw:: html

   <video width="660" src="https://media.defn.io/dramatiq-2.mp4" controls></video>

Here's what it looks like:

::

  import dramatiq
  import requests


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


  # Synchronously count the words on example.com in the current process
  count_words("http://example.com")

  # or send the actor a message so that it may perform the count
  # later, in a separate process.
  count_words.send("http://example.com")

**Dramatiq** is :doc:`licensed<license>` under the LGPL and it
officially supports Python 3.10 and later.


Get It Now
----------

If you want to use it with RabbitMQ_::

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

Or if you want to use it with Redis_::

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

Read the :doc:`motivation` behind it or the :doc:`guide` if you're
ready to get started.


User Guide
----------

This part of the documentation is focused primarily on teaching you
how to use Dramatiq.

.. toctree::
   :maxdepth: 2

   installation
   motivation
   guide
   best_practices
   troubleshooting
   advanced
   cookbook


API Reference
-------------

This part of the documentation is focused on detailing the various
bits and pieces of the Dramatiq developer interface.

.. toctree::
   :maxdepth: 2

   reference


Project Info
------------

.. toctree::
   :maxdepth: 1

   Source Code <https://github.com/Bogdanp/dramatiq>
   changelog
   PyPI <https://pypi.org/project/dramatiq/>
   Contributing <https://github.com/Bogdanp/dramatiq/blob/master/CONTRIBUTING.md>
   Discussion Board <https://groups.io/g/dramatiq-users>
   license

---

## File: docs/source/installation.rst

.. include:: global.rst

Installation
============

Dramatiq supports Python versions 3.10 and up and is installable via
`pip`_ or from source.


Via pip
-------

To install dramatiq, simply run the following command in a terminal::

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

RabbitMQ_ is the recommended message broker, but Dramatiq also
supports Redis_.

If you would like to use it with Redis_ then run::

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

If you don't have `pip`_ installed, check out `this guide`_.

Extra Requirements
^^^^^^^^^^^^^^^^^^

When installing the package via pip you can specify the following
extra requirements:

================  =======================================================================================
Name              Description
================  =======================================================================================
``all``           Installs all available extras.
``gevent``        Installs the required dependencies for running workers under gevent.
``memcached``     Installs the required dependencies for the Memcached rate limiter backend.
``prometheus``    Installs the required dependencies for the Prometheus metrics middleware.
``rabbitmq``      Installs the required dependencies for using Dramatiq with RabbitMQ.
``redis``         Installs the required dependencies for using Dramatiq with Redis.
``watch``         Installs the required dependencies for the ``--watch`` flag.  (Supported only on UNIX)
================  =======================================================================================

If you want to install Dramatiq with all available features, run::

  $ pip install -U 'dramatiq[all]'

Optional Requirements
^^^^^^^^^^^^^^^^^^^^^

If you're using Redis as your broker and aren't planning on using PyPy
then you should additionally install the ``hiredis`` package to get an
increase in throughput.


From Source
-----------

To install the latest development version of dramatiq from source,
clone the repo from `GitHub`_

::

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

then install it to your local site-packages by running

::

   $ python setup.py install

in the cloned directory.


.. _GitHub: https://github.com/Bogdanp/dramatiq
.. _pip: https://pip.pypa.io/en/stable/
.. _this guide: http://docs.python-guide.org/en/latest/starting/installation/

---

## File: docs/source/motivation.rst

.. include:: global.rst

Motivation
==========

Dramatiq's primary reason for being is the fact that I wanted a
distributed task queueing library that is simple and has sane defaults
for most SaaS workloads.  In that sense, it draws a lot of inspiration
from `GAE Push Queues`_ and Sidekiq_.

Dramatiq's driving principles are as follows:

* high reliability and performance
* simple and easy to understand core
* convention over configuration

If you've ever had to use Celery in anger, Dramatiq could be the tool
for you.


.. _GAE Push Queues: https://cloud.google.com/appengine/docs/standard/python/taskqueue/push/
.. _Sidekiq: http://sidekiq.org


Compared to *
-------------

.. note::
   This section was last updated in 2019.  It's possible that various
   bits of the table below are now outdated.  Clarifications in PR form
   are always welcome!

I've used Celery_ professionally for years and my growing frustration
with it is one of the reasons why I developed dramatiq.  Here are some
of the main differences between Dramatiq, Celery, Huey and RQ:

+------------------------------+----------+---------------+--------------+--------------+
|                              | Dramatiq | Celery_       | Huey_        | RQ_          |
+------------------------------+----------+---------------+--------------+--------------+
| Python 2 support             | No       | Yes           | Yes          | Yes          |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Windows support              | Yes      | No            | Yes          | No           |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Simple implementation        | Yes      | No [#sim]_    | Yes          | Yes          |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Automatic retries            | Yes      | No            | Yes          | No           |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Reliable delivery            | Yes      | Optional [#]_ | No           | No           |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Locks and rate limiting      | Yes      | No            | Yes          | No           |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Task prioritization          | Yes      | No [#prio]_   | Yes          | Yes          |
|                              | [#prio]_ |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Delayed tasks                | Yes      | Yes [#]_      | Yes          | No           |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Cronlike scheduling          | No       | Yes           | Yes          | No           |
|                              | [#cron]_ |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Chaining / Pipelining        | Yes      | Yes           | Yes          | No           |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Result storage               | Yes      | Yes           | Yes          | Yes          |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Code auto-reload             | Yes      | No            | No           | No           |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| RabbitMQ support             | Yes      | Yes           | Yes          | No           |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Redis support                | Yes      | Yes           | Yes          | Yes          |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| In-memory broker support     | Yes      | No            | Yes          | No           |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+
| Greenlet support             | Yes      | Yes           | Yes          | No           |
|                              |          |               |              |              |
+------------------------------+----------+---------------+--------------+--------------+

.. [#] Celery acks tasks as soon as they’re pulled by a worker by
       default.  This is easy to change, but a bad default.  Dramatiq
       doesn’t let you change this: tasks are only ever acked when
       they’re done processing.

.. [#] Celery has poor support for delayed tasks.  Delayed tasks are
       put on the same queue that is used for normal tasks and they’re
       simply pulled into worker memory until they can be executed,
       making it hard to autoscale workers by queue size.  Dramatiq
       enqueues delayed tasks on a separate queue and moves them back
       when they're ready to be executed.

.. [#sim] Celery's source code is spread across 3 different projects
          (celery, billiard and kombu) and it’s impenetrable.  Its
          usage of runtime stack frame manipulation leads to
          heisenbugs.

.. [#prio] Celery doesn't support task prioritization.  Dramatiq
           supports global prioritization under RabbitMQ via the
           ``broker_priority``.  It also provides worker-local
           prioritization of prefetched messages.

.. [#cron] For cron-like scheduling functionality, you can combine
           Dramatiq with APScheduler_ or Periodiq_. 3rd-party packages
           exist which implement this approach for various frameworks,
           like: Dramatiq-Crontab_


.. _Celery: http://celeryproject.org
.. _huey: https://huey.readthedocs.io/
.. _RQ: http://python-rq.org/
.. _APScheduler: https://apscheduler.readthedocs.io/
.. _Periodiq: https://gitlab.com/bersace/periodiq
.. _Dramatiq-Crontab: https://github.com/voiio/dramatiq-crontab