## File: docs/api.rst
.. _api:
Huey's API
==========
Most end-users will interact with the API using the two decorators:
* :py:meth:`Huey.task`
* :py:meth:`Huey.periodic_task`
The API documentation will follow the structure of the huey ``api.py`` module.
Huey types
----------
.. note::
See the documentation for :py:class:`Huey` for the list of initialization
parameters common to all Huey implementations.
.. py:class:: RedisHuey
:py:class:`Huey` that utilizes `redis `_ for queue and
result storage. Requires `redis-py `_.
Commonly-used keyword arguments for storage configuration:
:param bool blocking: Use blocking-pop when reading from the queue (as
opposed to polling). Default is true.
:param connection_pool: a redis-py ``ConnectionPool`` instance.
:param url: url for Redis connection.
:param host: hostname of the Redis server.
:param port: port number.
:param password: password for Redis.
:param int db: Redis database to use (typically 0-15, default is 0).
:param bool notify_result: use a blocking-pop on a result-ready key to
enable low-latency result reading.
:param int notify_result_ttl: TTL for result-ready key to automatically
expire un-awaited results.
The `redis-py documentation `_
contains the complete list of arguments supported by the Redis client.
.. note::
RedisHuey does not support task priorities. If you wish to use task
priorities with Redis, use :py:class:`PriorityRedisHuey`.
RedisHuey uses a Redis LIST to store the queue of pending tasks. Redis
lists are a natural fit, as they offer O(1) append and pop from either end
of the list. Redis also provides blocking-pop commands which allow the
consumer to react to a new message as soon as it is available without
resorting to polling.
For low-latency result-fetching, you can specify ``notify_result=True``
when instantiating ``RedisHuey`` (or subclasses). This works by
coordinating result readiness through a blocking pop on a dedicated result
list. By default this result-readiness list will expire after 86400 seconds,
but this can be controlled with the ``notify_result_ttl`` parameter.
**All RedisHuey** implementations support these options.
.. seealso:: :py:class:`RedisStorage`
.. py:class:: PriorityRedisHuey
:py:class:`Huey` that utilizes `redis `_ for queue and
result storage. Requires `redis-py `_.
Accepts the same arguments as :py:class:`RedisHuey`.
PriorityRedisHuey supports :ref:`task priorities `, and requires
Redis **5.0 or newer**.
PriorityRedisHuey uses a Redis SORTED SET to store the queue of pending
tasks. Sorted sets consist of a unique value and a numeric score. In
addition to being sorted by numeric score, Redis also orders the items
within the set lexicographically. Huey takes advantage of these two
characteristics to implement the priority queue. Redis 5.0 added a new
command, ZPOPMIN, which pops the lowest-scoring item from the sorted set
(and BZPOPMIN, the blocking variety).
.. py:class:: RedisExpireHuey
Identical to :py:class:`RedisHuey` except for the way task result values
are stored. RedisHuey keeps all task results in a Redis hash, and whenever
a task result is read (via the result handle), it is also removed from the
result hash. This is done to prevent the task result storage from growing
without bound. Additionally, using a Redis hash for all results helps avoid
cluttering up the Redis keyspace and utilizes less RAM for storing the keys
themselves.
``RedisExpireHuey`` uses a different approach: task results are stored in
ordinary Redis keys with a special prefix. Result keys are then given a
time-to-live, and will be expired automatically by the Redis server. This
removes the necessity to remove results from the result store after they
are read once.
Commonly-used keyword arguments for storage configuration:
:param int expire_time: Expire time in seconds, default is 86400 (1 day).
:param bool blocking: Use blocking-pop when reading from the queue (as
opposed to polling). Default is true.
:param connection_pool: a redis-py ``ConnectionPool`` instance.
:param url: url for Redis connection.
:param host: hostname of the Redis server.
:param port: port number.
:param password: password for Redis.
:param int db: Redis database to use (typically 0-15, default is 0).
:param bool notify_result: use a blocking-pop on a result-ready key to
enable low-latency result reading.
:param int notify_result_ttl: TTL for result-ready key to automatically
expire un-awaited results.
.. py:class:: PriorityRedisExpireHuey
Combines behavior of :py:class:`RedisExpireHuey` and
:py:class:`PriorityRedisHuey`.
.. py:class:: SqliteHuey
:py:class:`Huey` that utilizes sqlite3 for queue and result storage. Only
requirement is the standard library ``sqlite3`` module.
Commonly-used keyword arguments:
:param str filename: filename for database, defaults to 'huey.db'.
:param int cache_mb: megabytes of memory to allow for sqlite page-cache.
:param bool fsync: use durable writes. Slower but more resilient to
corruption in the event of sudden power loss. By default will use the
SQLite library's ``synchronous`` setting (typically ``1``, which
provides a good balance between durability and performance).
SqliteHuey fully supports task priorities.
.. seealso:: :py:class:`SqliteStorage`
.. py:class:: CySqliteHuey
:py:class:`Huey` that utilizes sqlite for queue and result storage, driven
by `cysqlite `_ rather than the standard
library ``sqlite3`` module (``pip install huey[cysqlite]``).
Commonly-used keyword arguments:
:param str filename: filename for database, defaults to 'huey.db'.
:param dict pragmas: mapping of pragmas to apply to every connection. If
unset, ``journal_mode`` will default to ``'wal'``, which enables
readers to coexist with a single writer.
CySqliteHuey fully supports task priorities.
.. seealso:: :py:class:`CySqliteStorage`
.. py:class:: PostgresHuey
:py:class:`Huey` that utilizes PostgreSQL for queue, schedule and result
storage. Requires `psycopg `_ 3.2 or newer
(``pip install huey[postgres]``).
Workers use ``LISTEN``/``NOTIFY`` for low-latency dequeue without
polling. The worker polls once before waiting and polls again when the
wait times out, so a missed notification delays a task by at most
``read_timeout`` seconds and can never lose one.
Commonly-used keyword arguments for storage configuration:
:param str dsn: connection string or URL, e.g.
``postgresql://user:pass@host:5432/dbname``. Alternatively, discrete
connection parameters (``host``, ``port``, ``dbname``, ``user``,
``password``, etc.) may be given as additional keyword arguments,
which are passed directly to ``psycopg.connect()``.
:param connection: zero-argument callable returning a new ``psycopg``
connection, to integrate with the application's own connection
machinery. Huey manages the lifetime of the connections it obtains.
:param bool blocking: wake workers using LISTEN/NOTIFY (default true).
Set to false to fall back to polling, e.g. when connecting through
PgBouncer in transaction-pooling mode, which breaks LISTEN.
:param read_timeout: maximum seconds a blocking dequeue waits before the
worker loops and polls again. Default is 1 second.
:param str table_prefix: prefix for the four tables huey creates
(``_task``, ``_schedule``, ``_kv``,
``_counter``). Default is ``huey``. Tables are created
automatically if they do not exist.
Each worker thread or process maintains one additional connection used
for LISTEN, so a consumer with N workers holds N+1 connections.
PostgresHuey fully supports task priorities.
.. seealso:: :py:class:`PostgresStorage`
.. py:class:: MemoryHuey
:py:class:`Huey` that uses in-memory storage. Only should be used when
testing or when using ``immediate`` mode. MemoryHuey fully supports task
priorities.
.. py:class:: FileHuey
:py:class:`Huey` that uses the file-system for storage. Should not be used
in high-throughput, highly-concurrent environments, as the
:py:class:`FileStorage` utilizes exclusive locks around all file-system
operations.
:param str path: base-path for huey data (queue tasks, schedule and results
will be stored in sub-directories of this path).
:param int levels: number of levels in result-file directory structure to
ensure the results directory does not contain an unmanageable number of
files.
:param bool use_thread_lock: use the standard lib ``threading.Lock``
instead of a lockfile for file-system operations. This should only be
enabled when using the greenlet or thread consumer worker models.
FileHuey fully supports task priorities.
.. py:class:: BlackHoleHuey
:py:class:`Huey` that uses :py:class:`BlackHoleStorage`. All storage
operations are no-ops: enqueued tasks are silently discarded and the result
store is always empty. Intended for testing scenarios where you want to
verify that tasks are being called without actually running them or storing
anything.
Huey object
-----------
.. py:class:: Huey(name='huey', results=True, store_none=False, utc=True, immediate=False, serializer=None, compression=False, use_zlib=False, immediate_use_memory=True, store_intermediate_errors=True, storage_kwargs)
:param str name: the name of the task queue, e.g. your application's name.
:param bool results: whether to store task results.
:param bool store_none: whether to store ``None`` in the result store.
:param bool store_intermediate_errors: when a task fails but has retries
remaining, store the intermediate exception in the result store and run
any ``on_error`` handler. When ``False``, the error is withheld until the
task's retries are exhausted. Defaults to ``True`` for backwards
compatibility. See :ref:`store-intermediate-errors`.
:param bool utc: use UTC internally, convert naive datetimes from local
time to UTC (if local time is other than UTC).
:param bool immediate: useful for debugging, causes tasks to be executed
synchronously in the application.
:param Serializer serializer: serializer implementation for tasks and
result data. The default implementation uses ``pickle``.
:param bool compression: compress tasks and result data.
:param bool use_zlib: use zlib for compression instead of gzip.
:param bool immediate_use_memory: automatically switch to a local in-memory
storage backend whenever immediate-mode is enabled.
:param storage_kwargs: arbitrary keyword arguments that will be passed to
the storage backend for additional configuration.
Huey executes tasks by exposing function decorators that cause the function
call to be enqueued for execution by the consumer.
Typically your application will only need one Huey instance, but you can
have as many as you like. The only caveat is that one consumer process
must be executed for each Huey instance.
Example usage:
.. code-block:: python
# demo.py
from huey import RedisHuey
# Create a huey instance.
huey = RedisHuey('my-app')
@huey.task()
def add_numbers(a, b):
return a + b
@huey.periodic_task(crontab(minute='0', hour='2'))
def nightly_report():
generate_nightly_report()
To run the consumer with 4 worker threads:
.. code-block:: console
$ huey_consumer demo.huey -w 4
To add two numbers, the "huey" way:
.. code-block:: pycon
>>> from demo import add_numbers
>>> res = add_numbers(1, 2)
>>> res(blocking=True) # Blocks until result is available.
3
To test huey without using a consumer, you can use "immediate" mode.
Immediate mode follows all the same code paths as Huey does when running
the consumer process, but does so synchronously within the application.
.. code-block:: pycon
>>> from demo import add_numbers, huey
>>> huey.immediate = True # Tasks executed immediately.
>>> res = add_numbers(2, 3)
>>> res()
5
.. py:attribute:: immediate
The ``immediate`` property is used to enable and disable :ref:`immediate mode `.
When immediate mode is enabled, task-decorated functions are executed
synchronously by the caller, making it very useful for development and
testing. Calling a task function still returns a :py:class:`Result`
handle, but the task itself is executed immediately.
By default, when immediate mode is enabled, Huey will switch to using
in-memory storage. This is to help prevent accidentally writing to a
live Redis server while testing. To disable this functionality, specify
``immediate_use_memory=False`` when initializing :py:class:`Huey`.
Enabling immediate mode:
.. code-block:: python
huey = RedisHuey()
# Enable immediate mode. Tasks now executed synchronously.
# Additionally, huey will now use in-memory storage.
huey.immediate = True
# Disable immediate mode. Tasks will now be enqueued in a Redis
# queue.
huey.immediate = False
Immediate mode can also be specified when your Huey instance is
created:
.. code-block:: python
huey = RedisHuey(immediate=True)
.. py:method:: create_consumer(**options)
:param options: keyword arguments passed to the :py:class:`Consumer`
constructor (e.g. ``workers``, ``worker_type``, ``periodic``,
``initial_delay``, etc.). See :ref:`consumer-options`.
:returns: a :py:class:`Consumer` instance.
Create a consumer programmatically, rather than using the
``huey_consumer`` command-line tool. This is useful for embedding
the consumer within your own application process, or for advanced
testing scenarios.
.. code-block:: python
consumer = huey.create_consumer(workers=4, worker_type='thread')
consumer.start()
# ... later ...
consumer.stop(graceful=True)
.. py:method:: task(retries=0, retry_delay=0, retry_backoff=0, priority=None, context=False, name=None, expires=None, timeout=None, **kwargs)
:param int retries: number of times to retry the function if an
unhandled exception occurs when it is executed.
:param int retry_delay: number of seconds to wait between retries.
:param retry_backoff: multiplier applied to the delay after each failed
attempt, giving exponentially-growing waits between retries. The
first retry waits ``retry_delay`` seconds. Defaults to ``0``, which
disables backoff. A value of ``1`` leaves the delay constant, and
values below ``1`` shrink it. Floats are accepted. See
:ref:`recipe-exponential-backoff`.
:param int priority: priority assigned to task, higher numbers are
processed first by the consumer when there is a backlog. See
:ref:`priority` (supported by every storage backend except
:py:class:`RedisHuey` and :py:class:`RedisExpireHuey`, which have
priority-aware counterparts).
:param bool context: when the task is executed, include the
:py:class:`Task` instance as a keyword argument.
:param str name: name for this task. If not provided, Huey will default
to using the module name plus function name.
:param expires: set expiration time for task. If the task is not run
before ``expires``, it will be discarded. The ``expires`` parameter
can be either an integer (seconds), a timedelta, or a datetime. For
relative expiration values, the expire time will be resolved when
the task is enqueued.
:param int timeout: set execution timeout for task. Implementation
used by the consumer is specific to the worker type and is NOT
available for threads. Processes use SIGALRM, greenlet uses
``gevent.Timeout``. When using with threads it is necessary to use
cooperative timeout checking. See :ref:`cooperative-timeout` for
example.
:param kwargs: arbitrary key/value arguments that are passed to the
:py:class:`TaskWrapper` instance.
:returns: a :py:class:`TaskWrapper` that wraps the decorated function
and exposes a number of APIs for enqueueing the task.
Function decorator that marks the decorated function for processing by
the consumer. Calls to the decorated function will do the following:
1. Serialize the function call into a :py:class:`Task` suitable for
storing in the queue.
2. Enqueue the task for execution by the consumer.
3. Return a :py:class:`Result` handle, which can be used to check the
result of the task function, revoke the task (assuming it hasn't
started yet), reschedule the task, and more.
The ``task()`` decorator returns a :py:class:`TaskWrapper`, which
implements special methods for enqueueing the decorated function,
:py:meth:`~TaskWrapper.schedule` it for the future, chain tasks to
form a pipeline, and more.
Example task:
.. code-block:: python
@huey.task()
def slow(n):
time.sleep(n)
return n
Calling the ``slow()`` task will return immediately. We can block until
the task finishes by waiting for the result:
.. code-block:: pycon
>>> res = slow(10) # Returns immediately.
>>> res() # Non-blocking read - returns None if not ready.
>>> res(blocking=True) # Block until task finishes, ~10s.
10
When ``context=True``, the :py:class:`Task` instance is passed to
your function as a ``task`` keyword argument. This gives access to
the task ID, remaining retries, and :ref:`cooperative timeout
` APIs.
When the result store is disabled (``results=False``), calling a
task-decorated function returns ``None`` instead of a
:py:class:`Result` handle.
If you stack additional decorators on a task function, note that
decorators **above** ``@huey.task()`` run in the calling process,
while decorators **below** it run in the consumer worker.
For a thorough walkthrough with examples, see the :ref:`guide`. For
the full :py:class:`TaskWrapper` API (scheduling, pipelines,
revocation, etc.), see :py:class:`TaskWrapper`, :py:class:`Task`,
and :py:class:`Result`.
.. py:method:: periodic_task(validate_datetime, retries=0, retry_delay=0, retry_backoff=0, priority=None, context=False, name=None, expires=None, timeout=None, **kwargs)
:param function validate_datetime: function which accepts a
``datetime`` instance and returns whether the task should be
executed at the given time.
:param int retries: number of times to retry the function if an
unhandled exception occurs when it is executed.
:param int retry_delay: number of seconds to wait in-between retries.
:param retry_backoff: multiplier applied to the delay after each failed
attempt. See :py:meth:`~Huey.task`.
:param int priority: priority assigned to task, higher numbers are
processed first by the consumer when there is a backlog. See
:ref:`priority` (supported by every storage backend except
:py:class:`RedisHuey` and :py:class:`RedisExpireHuey`, which have
priority-aware counterparts).
:param bool context: when the task is executed, include the
:py:class:`Task` instance as a parameter.
:param str name: name for this task. If not provided, Huey will default
to using the module name plus function name.
:param expires: set expiration time for task. If the task is not run
before ``expires``, it will be discarded. The ``expires`` parameter
can be either an integer (seconds), a timedelta, or a datetime. For
relative expiration values, the expire time will be resolved when
the task is enqueued.
:param int timeout: set execution timeout for task. Implementation
used by the consumer is specific to the worker type and is NOT
available for threads. Processes use SIGALRM, greenlet uses
``gevent.Timeout``. When using with threads it is necessary to use
cooperative timeout checking. See :ref:`cooperative-timeout` for
example.
:param kwargs: arbitrary key/value arguments that are passed to the
:py:class:`TaskWrapper` instance.
:returns: a :py:class:`TaskWrapper` that wraps the decorated function
and exposes a number of APIs for enqueueing the task.
The ``periodic_task()`` decorator marks a function for automatic
execution by the consumer *at a specific interval*, like ``cron``.
The ``validate_datetime`` parameter is typically the :py:func:`crontab`
helper (described below). The consumer checks once per minute whether
the function should run.
Example, executing every three hours, on the hour:
.. code-block:: python
@huey.periodic_task(crontab(minute='0', hour='*/3'))
def update_feeds():
for feed in my_list_of_feeds:
fetch_feed_data(feed)
.. note::
Because functions decorated with ``periodic_task`` are meant to be
executed at intervals in isolation, they should not take any
required parameters nor should they be expected to return a
meaningful value.
Like :py:meth:`~Huey.task`, the periodic task decorator adds helpers
to the decorated function. These helpers allow you to
:py:meth:`~TaskWrapper.revoke` and :py:meth:`~TaskWrapper.restore` the
periodic task, enabling you to pause it or prevent its execution. For
more information, see :py:class:`TaskWrapper`.
.. note::
The result (return value) of a periodic task is not stored in the
result store. This is primarily due to the fact that there is not
an obvious way one would read such results, since the invocation of
the periodic task happens inside the consumer scheduler. As such,
there is no task result handle which the user could use to read the
result. To store the results of periodic tasks, you will need to
use your own storage or use the storage APIs directly:
.. code-block:: python
@huey.periodic_task(crontab(minute='*/10'))
def my_task():
# do some work...
do_something()
# Manually store some data in the result store.
huey.put('my-task', some_data_to_store)
More info:
* :py:meth:`Huey.put`
* :py:meth:`Huey.get`
.. py:method:: context_task(obj, as_argument=False, **kwargs)
:param obj: object that implements the context-manager APIs.
:param bool as_argument: pass the return value of the context-manager's
``__enter__`` method into the decorated task as the first argument.
:param kwargs: additional keyword arguments are forwarded to
:py:meth:`~Huey.task` (e.g. ``retries``, ``retry_delay``,
``priority``, ``context``, ``name``, ``expires``, ``timeout``).
:returns: a :py:class:`TaskWrapper` that wraps the decorated function
and exposes a number of APIs for enqueueing the task.
This is an extended implementation of the :py:meth:`Huey.task`
decorator, which wraps the decorated task in a ``with obj:`` block.
Roughly equivalent to:
.. code-block:: python
db = PostgresqlDatabase(...)
@huey.task()
def without_context_task(n):
with db:
do_something(n)
@huey.context_task(db)
def with_context_task(n):
return do_something(n)
When ``as_argument=True``, the value returned by ``__enter__`` is
passed as the first positional argument to the task function:
.. code-block:: python
@huey.context_task(db, as_argument=True)
def with_ctx(conn, n):
# conn is the value returned by db.__enter__()
return conn.execute(n)
.. py:method:: pre_execute(name=None)
:param name: (optional) name for the hook.
:returns: a decorator used to wrap the actual pre-execute function.
Decorator for registering a pre-execute hook. The callback will be
executed before the execution of every task. Execution of the task can
be cancelled by raising a :py:class:`CancelExecution` exception.
Uncaught exceptions will be logged but will not cause the task itself
to be cancelled.
The callback function should accept a single task instance, the return
value is ignored.
Hooks are executed in the order in which they are registered.
Usage:
.. code-block:: python
@huey.pre_execute()
def my_pre_execute_hook(task):
if datetime.datetime.now().weekday() == 6:
raise CancelExecution('Sunday, no work will be done.')
.. py:method:: unregister_pre_execute(name_or_fn)
:param name_or_fn: the name given to the pre-execute hook, or the
function object itself.
:returns: boolean
Unregister the specified pre-execute hook.
.. py:method:: post_execute(name=None)
:param name: (optional) name for the hook.
:returns: a decorator used to wrap the actual post-execute function.
Register a post-execute hook. The callback will be executed after the
execution of every task. Uncaught exceptions will be logged but will
have no other effect on the overall operation of the consumer.
The callback function should accept:
* a :py:class:`Task` instance
* the return value from the execution of the task (which may be None)
* any exception that was raised during the execution of the task (which
will be None for tasks that executed normally).
The return value of the callback itself is ignored.
Hooks are executed in the order in which they are registered.
Usage:
.. code-block:: python
@huey.post_execute()
def my_post_execute_hook(task, task_value, exc):
do_something()
.. py:method:: unregister_post_execute(name_or_fn)
:param name_or_fn: the name given to the post-execute hook, or the
function object itself.
:returns: boolean
Unregister the specified post-execute hook.
.. py:method:: on_startup(name=None)
:param name: (optional) name for the hook.
:returns: a decorator used to wrap the actual on-startup function.
Register a startup hook. The callback will be executed whenever a
worker comes online. Uncaught exceptions will be logged but will
have no other effect on the overall operation of the worker.
The callback function must not accept any parameters.
This API is provided to simplify setting up shared resources that, for
whatever reason, should not be created as import-time side-effects. For
example, your tasks need to write data into a Postgres database. If you
create the connection at import-time, before the worker processes are
spawned, you'll likely run into errors when attempting to use the
connection from the child (worker) processes. To avoid this problem,
you can register a startup hook which executes once when the worker
starts up.
Usage:
.. code-block:: python
db_connection = None
@huey.on_startup()
def setup_db_connection():
global db_connection
db_connection = psycopg2.connect(database='my_db')
@huey.task()
def write_data(rows):
cursor = db_connection.cursor()
# ...
.. py:method:: unregister_on_startup(name_or_fn)
:param name_or_fn: the name given to the on-startup hook, or the
function object itself.
:returns: boolean
Unregister the specified on-startup hook.
.. py:method:: on_shutdown(name=None)
:param name: (optional) name for the hook.
:returns: a decorator used to wrap the actual on-shutdown function.
Register a shutdown hook. The callback will be executed by a worker
immediately before it goes offline. Uncaught exceptions will be logged
but will have no other effect on the overall shutdown of the worker.
The callback function must not accept any parameters.
This API is provided to simplify cleaning-up shared resources.
.. py:method:: unregister_on_shutdown(name_or_fn)
:param name_or_fn: the name given to the on-shutdown hook, or the
function object itself.
:returns: boolean
Unregister the specified on-shutdown hook.
.. py:method:: signal(*signals)
:param signals: zero or more signals to handle.
:returns: a decorator used to wrap the actual signal handler.
Attach a signal handler callback, which will be executed when the
specified signals are sent by the consumer. If no signals are listed,
then the handler will be bound to **all** signals. The list of signals
and additional information can be found in the :ref:`signals`
documentation.
Example:
.. code-block:: python
from huey.signals import SIGNAL_ERROR, SIGNAL_LOCKED
@huey.signal(SIGNAL_ERROR, SIGNAL_LOCKED)
def task_not_run_handler(signal, task, exc=None):
# Do something in response to the "ERROR" or "LOCKED" signals.
# Note that the "ERROR" signal includes a third parameter,
# which is the unhandled exception that was raised by the task.
# Since this parameter is not sent with the "LOCKED" signal, we
# provide a default of ``exc=None``.
pass
.. py:method:: disconnect_signal(receiver, *signals)
:param receiver: the signal handling function to disconnect.
:param signals: zero or more signals to stop handling.
Disconnect the signal handler from the provided signals. If no signals
are provided, then the handler is disconnected from any signals it may
have been connected to.
.. py:method:: notify_interrupted_tasks()
Emit ``SIGNAL_INTERRUPTED`` for every task that is currently being
executed (i.e. in-flight). This is called automatically by the consumer
during shutdown. Application code should not normally need to call this
method directly, but it is available for custom consumer implementations
or advanced testing.
Tasks are tracked from the moment execution begins until execution
completes (or fails). If the consumer is killed mid-task, those tasks
remain in the ``_tasks_in_flight`` set, and this method emits
``SIGNAL_INTERRUPTED`` for each one.
.. py:method:: build_error_result(task, exception)
:param task: the :py:class:`Task` that raised.
:param Exception exception: the unhandled exception.
:returns: dict stored in the result store as the task's error result.
Build the metadata stored when a task fails. The default implementation
returns the exception's ``repr()``, the formatted traceback, the retry
count and the task id. Override it on a :py:class:`Huey` subclass to
record additional context. See :ref:`recipe-custom-error-metadata`.
.. py:method:: enqueue(task)
:param task: a :py:class:`Task` instance, a :py:class:`group`, or a
:py:class:`chord`.
:returns: depends on the input type (see below).
Enqueue the given task, group, or chord for execution. When the result
store is enabled (default), the return value depends on what was
enqueued:
* A single :py:class:`Task`: returns a :py:class:`Result` handle.
* A :py:class:`Task` with a pipeline (via :py:meth:`Task.then`):
returns a :py:class:`ResultGroup` containing a :py:class:`Result`
for each task in the pipeline.
* A :py:class:`group`: returns a :py:class:`ResultGroup` containing a
:py:class:`Result` for each member task.
* A :py:class:`chord`: returns a :py:class:`ChordResult`, which
provides access to the member results, the callback result, and any
tasks chained after the callback.
.. note::
Calling (or scheduling) a ``task``-decorated function will
automatically enqueue a task for execution. You only need to call
:py:meth:`~Huey.enqueue` directly when working with pipelines,
groups, or chords.
.. py:method:: revoke(task, revoke_until=None, revoke_once=False)
.. seealso:: Use :py:meth:`Result.revoke` instead.
.. py:method:: revoke_by_id(task_id, revoke_until=None, revoke_once=False)
:param str task_id: task instance id.
:param datetime revoke_until: optional expiration date for revocation.
:param bool revoke_once: revoke once and then re-enable.
Revoke a :py:class:`Task` instance using the task id.
.. py:method:: revoke_all(task_class, revoke_until=None, revoke_once=False)
.. seealso:: Use :py:meth:`TaskWrapper.revoke` instead.
.. py:method:: restore(task)
.. seealso:: Use :py:meth:`Result.restore` instead.
.. py:method:: restore_by_id(task_id)
:param str task_id: task instance id.
:returns: boolean indicating success.
Restore a :py:class:`Task` instance using the task id. Returns boolean
indicating if the revocation was successfully removed.
.. py:method:: restore_all(task_class)
.. seealso:: Use :py:meth:`TaskWrapper.restore` instead.
.. py:method:: is_revoked(task, timestamp=None)
:param task: either a task instance, a task ID, a Result, or a Task class.
This method should rarely need to be called directly. Typically you
should rather use the ``is_revoked`` method on the object that is being
revoked, for example:
.. code-block:: python
@huey.task()
def greet(name):
return 'Hello %s' % name
r = greet.schedule(delay=60, args=('Huey',))
r.revoke() # Revoke this task.
r.is_revoked() # True.
greet.revoke() # Revoke ALL invocations of this task.
greet.is_revoked() # True.
.. seealso::
For task instances, use :py:meth:`Result.is_revoked`.
For task functions, use :py:meth:`TaskWrapper.is_revoked`.
.. py:method:: result(task_id, blocking=False, timeout=None, backoff=1.15, max_delay=1.0, revoke_on_timeout=False, preserve=False)
:param task_id: the task's unique identifier.
:param bool blocking: whether to block while waiting for task result
:param timeout: number of seconds to block (if ``blocking=True``)
:param backoff: amount to backoff delay each iteration of loop
:param max_delay: maximum amount of time to wait between iterations when
attempting to fetch result.
:param bool revoke_on_timeout: if a timeout occurs, revoke the task,
thereby preventing it from running if it is has not started yet.
:param bool preserve: when set to ``True``, this parameter ensures that
the task result will be preserved after having been successfully
retrieved. Ordinarily, Huey will discard results after they have
been read, to prevent the result store from growing without bounds.
Attempts to retrieve the return value of a task. By default, :py:meth:`~Huey.result`
will check for the value, returning ``None`` if it is not ready
yet. If you want to wait for the result, specify ``blocking=True``.
This will loop, backing off up to the provided ``max_delay``, until the
value is ready or the ``timeout`` is reached. If the ``timeout`` is
reached before the result is ready, a :py:class:`ResultTimeout` will be
raised.
.. seealso::
:py:class:`Result`. The :py:meth:`~Huey.result` method is a
wrapper that creates a ``Result`` object and calls its
:py:meth:`~Result.get` method.
.. note:: If the task failed with an exception, then a
:py:class:`TaskException` that wraps the original exception will be
raised.
.. warning:: By default the result store will delete a task's return
value after the value has been successfully read (by a successful
call to the :py:meth:`~Huey.result` or :py:meth:`Result.get`
methods). If you intend to access the task result multiple times,
you must specify ``preserve=True`` when calling these methods.
.. py:method:: lock_task(lock_name)
:param str lock_name: Name to use for the lock.
:returns: :py:class:`TaskLock` instance, which can be used as a
decorator or context-manager.
Utilize the Storage key/value APIs to implement simple locking.
This lock is designed to be used to prevent multiple invocations of a
task from running concurrently. Can be used as either a context-manager
within the task, or as a task decorator. If using as a decorator, place
it directly above the function declaration.
If a second invocation occurs and the lock cannot be acquired, then a
:py:class:`TaskLockedException` is raised, which is handled by the
consumer. The task will not be executed and a ``SIGNAL_LOCKED`` will be
sent. If the task is configured to be retried, then it will be retried
normally.
Examples:
.. code-block:: python
@huey.periodic_task(crontab(minute='*/5'))
@huey.lock_task('reports-lock') # Goes *after* the task decorator.
def generate_report():
# If a report takes longer than 5 minutes to generate, we do
# not want to kick off another until the previous invocation
# has finished.
run_report()
@huey.periodic_task(crontab(minute='0'))
def backup():
# Generate backup of code
do_code_backup()
# Generate database backup. Since this may take longer than an
# hour, we want to ensure that it is not run concurrently.
with huey.lock_task('db-backup'):
do_db_backup()
.. py:method:: is_locked(lock_name)
:param str lock_name: Name of lock to check.
:returns: boolean value indicating whether lock is held or not.
.. py:method:: flush_locks(*names)
:param names: optional additional lock-names to flush.
:returns: set of lock names that were held and subsequently cleared.
Flush any locks that may be held. Locks created by using
:py:meth:`~Huey.lock_task` as a **decorator** are automatically
discovered (registered at import time). Locks created inside a
function body using the **context-manager** form are not automatically
discovered, and their names must be passed explicitly.
When ``names`` are provided, both the discovered locks **and** the
explicitly named locks are flushed.
.. code-block:: python
# Decorator form, automatically discovered:
@huey.periodic_task(crontab(minute='*/5'))
@huey.lock_task('reports-lock')
def generate_report():
run_report()
# Context-manager form, NOT automatically discovered:
@huey.task()
def backup():
with huey.lock_task('db-backup'):
do_db_backup()
# Flush discovered locks only:
huey.flush_locks() # Clears 'reports-lock' if held.
# Flush discovered locks AND the context-manager lock:
huey.flush_locks('db-backup') # Clears both.
.. py:method:: rate_limit(name, limit, per, retry=True)
:param str name: Name to use for the rate-limiter.
:param int limit: Maximum invocations in the given window.
:param int per: Size of rate-limiting window in seconds.
:param bool retry: Automatically reschedule rate-limited tasks to run
at beginning of next rate-limit window.
:returns: :py:class:`RateLimit` instance, which can be used as a
decorator or context-manager.
Simple fixed-window rate-limiting for tasks.
When a task fails due to being rate-limited, it will be automatically
rescheduled to run at the start of the next window when ``retry=True`` or
the task itself has one or more ``retries``.
Tasks that fail due to rate-limiting will emit a ``SIGNAL_RATE_LIMITED``.
By default any task that is rate-limited will be retried at the start
of the next rate-limiting window. For instance say we have a task that
is limited to 10 calls / 60 seconds. The 11th call in a given 60s
period will be retried at the start of the next window (~1 to 60 seconds
later, depending on timing).
This is good for tasks that must not be lost. There is a risk if the
volume of tasks continually exceeds the rate-limit that the automatic
retries will grow continuously. To mitigate this you can specify the
number of retries in the task decorator and disable automatic retries
on the rate limiter. See examples below.
Example of limiting to 10 calls per minute. Tasks that are limited will
automatically be retried at the start of the next rate-limit window.
.. code-block:: python
@huey.task()
@huey.rate_limit('data_sync', limit=10, per=60)
def data_sync(data):
service.apply_updates(data)
In the above example, if a task fails due to being rate-limited, it
will be rescheduled for the start of the next window. If it fails
again due to rate-limiting, it will again be rescheduled, etc.
Example of limiting to 10 calls per minute AND only allowing 2 retries.
This mitigates unbounded growth for rate-limits that are continuously
being hit:
.. code-block:: python
@huey.task(retries=2)
@huey.rate_limit('data_sync', limit=10, per=60)
def data_sync(data):
service.apply_updates(data)
In the above example, if a task fails due to being rate-limited, it
will be rescheduled for the start of the next window up to 2 times.
Equivalent example using a context-manager instead of a decorator:
.. code-block:: python
@huey.task(retries=2)
def data_sync(data):
with huey.rate_limit('data_sync', limit=10, per=60):
service.apply_updates(data)
By default, rate-limited tasks are scheduled to be retried at the
beginning of the next rate-limit window. If a task specifies an
explicit ``retry_delay``, however, that value will be used instead:
.. code-block:: python
@huey.task(retries=2, retry_delay=120)
@huey.rate_limit('data_sync', limit=10, per=60)
def data_sync(data):
service.apply_updates(data)
In the above code, if the task fails due to being rate-limited, it will
be retried up to 2 times after a delay of 120s between retries.
.. py:method:: put(key, value)
:param key: key for data
:param value: arbitrary data to store in result store.
Store a value in the result-store under the given key.
.. py:method:: get(key, peek=False)
:param key: key to read
:param bool peek: non-destructive read
Read a value from the result-store at the given key. By default reads
are destructive. To preserve the value for additional reads, specify
``peek=True``.
.. py:method:: delete(key)
:param key: key to delete from the result store.
:returns: boolean indicating whether the key existed and was deleted.
Remove a value from the result-store at the given key. Useful for
cleaning up manually-stored data created with :py:meth:`~Huey.put`.
.. py:method:: put_if_empty(key, value)
:param key: key to store data under.
:param value: arbitrary data to store.
:returns: boolean indicating whether the value was stored. Returns
``False`` if the key already exists.
Atomically store a value only if the key does not already exist. This
is the primitive used internally by :py:class:`TaskLock` to implement
exclusive locking. It can also be used to build custom coordination
patterns like task deduplication.
.. py:method:: pending(limit=None)
:param int limit: optionally limit the number of tasks returned.
:returns: a list of :py:class:`Task` instances waiting to be run.
.. note::
This method must deserialize every task in the queue, which can be
slow if the queue is large. Use :py:meth:`~Huey.pending_count` when
you only need the count.
.. py:method:: pending_count()
:returns: the number of tasks currently in the queue, without
deserializing them.
.. py:method:: scheduled(limit=None)
:param int limit: optionally limit the number of tasks returned.
:returns: a list of :py:class:`Task` instances that are scheduled to
execute at some time in the future.
.. note::
This method deserializes every task on the schedule. Use
:py:meth:`~Huey.scheduled_count` when you only need the count.
.. py:method:: scheduled_count()
:returns: the number of tasks currently in the schedule, without
deserializing them.
.. py:method:: all_results()
:returns: a dict of task-id to the serialized result data for all
key/value pairs in the result store.
.. py:method:: result_count()
:returns: the number of key/value pairs in the result store.
.. py:method:: flush()
Remove all data from the queue, schedule, and result store. This is
a destructive operation. Primarily useful for testing.
.. py:method:: __len__()
Return the number of items currently in the queue.
.. py:class:: TaskWrapper(huey, func, retries=None, retry_delay=None, context=False, name=None, task_base=None, **settings)
:param Huey huey: A huey instance.
:param func: User function.
:param int retries: Upon failure, number of times to retry the task.
:param int retry_delay: Number of seconds to wait before retrying after a
failure/exception.
:param bool context: when the task is executed, include the
:py:class:`Task` instance as a parameter.
:param str name: Name for task (will be determined based on task module and
function name if not provided).
:param task_base: Base-class for task, defaults to :py:class:`Task`.
:param settings: Arbitrary settings to pass to the task class constructor.
Wrapper around a user-defined function that converts function calls into
tasks executed by the consumer. The wrapper, which decorates the function,
replaces the function in the scope with a :py:class:`TaskWrapper` instance.
The wrapper class, when called, will enqueue the requested function call
for execution by the consumer.
.. note::
You should not need to create :py:class:`TaskWrapper` instances
directly. The :py:meth:`Huey.task` and :py:meth:`Huey.periodic_task`
decorators will create the appropriate TaskWrapper automatically.
.. py:method:: schedule(args=None, kwargs=None, eta=None, delay=None, priority=None, retries=None, retry_delay=None, retry_backoff=None, expires=None, timeout=None, id=None)
:param tuple args: arguments for decorated function.
:param dict kwargs: keyword arguments for decorated function.
:param datetime eta: the time at which the function should be executed.
:param int delay: number of seconds to wait before executing function.
:param int priority: priority assigned to task, higher numbers are
processed first by the consumer when there is a backlog.
:param int retries: number of times to retry the function if an
unhandled exception occurs when it is executed.
:param int retry_delay: number of seconds to wait between retries.
:param retry_backoff: multiplier applied to the delay after each failed
attempt. See :py:meth:`~Huey.task`.
:param expires: set expiration time for task. If the task is not run
before ``expires``, it will be discarded. The ``expires`` parameter
can be either an integer (seconds), a timedelta, or a datetime. For
relative expiration values, the expire time will be resolved when
the task is enqueued.
:param int timeout: set execution timeout for task. Implementation
used by the consumer is specific to the worker type and is NOT
available for threads. Processes use SIGALRM, greenlet uses
``gevent.Timeout``. When using with threads it is necessary to use
cooperative timeout checking. See :ref:`cooperative-timeout` for
example.
:param id: assign the given ``id`` to the Task being scheduled.
:returns: a :py:class:`Result` handle for the task.
Use the ``schedule`` method to schedule the execution of the queue task
for a given time in the future:
.. code-block:: python
import datetime
one_hour = datetime.datetime.now() + datetime.timedelta(hours=1)
# Schedule the task to be run in an hour. It will be called with
# three arguments.
res = check_feeds.schedule(args=(url1, url2, url3), eta=one_hour)
# Equivalent, but uses delay rather than eta.
res = check_feeds.schedule(args=(url1, url2, url3), delay=3600)
.. py:method:: revoke(revoke_until=None, revoke_once=False)
:param datetime revoke_until: Automatically restore the task after the
given datetime.
:param bool revoke_once: Revoke the next execution of the task and then
automatically restore.
Revoking a task will prevent any instance of the given task from
executing. When no parameters are provided the function will not
execute again until :py:meth:`TaskWrapper.restore` is called.
This function can be called multiple times, but each call will
supersede any restrictions from the previous revocation.
.. code-block:: python
# Skip the next execution
send_emails.revoke(revoke_once=True)
# Prevent any invocation from executing.
send_emails.revoke()
# Prevent any invocation for 24 hours.
tomorrow = datetime.datetime.now() + datetime.timedelta(days=1)
send_emails.revoke(revoke_until=tomorrow)
.. py:method:: is_revoked(timestamp=None)
:param datetime timestamp: If provided, checks whether task is revoked
with respect to the given timestamp.
:returns: bool indicating whether task is revoked.
Check whether the given task is revoked.
.. py:method:: restore()
:returns: bool indicating whether a previous revocation rule was found
and removed successfully.
Removes a previous task revocation, if one was configured.
.. py:method:: call_local()
Call the ``@task``-decorated function, bypassing all Huey-specific
logic. In other words, ``call_local()`` provides access to the
underlying user-defined function.
.. code-block:: pycon
>>> add.call_local(1, 2)
3
.. py:method:: s(*args, **kwargs)
:param args: Arguments for task function.
:param kwargs: Keyword arguments for task function.
:param int priority: assign priority override to task, higher numbers
are processed first by the consumer when there is a backlog.
:param expires: set expiration time for task. If the task is not run
before ``expires``, it will be discarded. The ``expires`` parameter
can be either an integer (seconds), a timedelta, or a datetime. For
relative expiration values, the expire time will be resolved when
the task is enqueued.
:param int timeout: set execution timeout for task. Implementation
used by the consumer is specific to the worker type and is NOT
available for threads. Processes use SIGALRM, greenlet uses
``gevent.Timeout``. When using with threads it is necessary to use
cooperative timeout checking. See :ref:`cooperative-timeout` for
example.
:returns: a :py:class:`Task` instance representing the execution of the
task function with the given arguments.
Create a :py:class:`Task` instance representing the invocation of the
task function with the given arguments and keyword-arguments.
.. note:: The returned task instance is **not** enqueued automatically.
.. warning::
The following keyword argument names are reserved and will be
intercepted by Huey rather than passed to your task function:
``eta``, ``delay``, ``retries``, ``retry_delay``, ``priority``,
``expires``, and ``timeout``. If your task function has a parameter
with one of these names, rename the parameter or pass the value
positionally.
To illustrate the distinction, when you call a ``task()``-decorated
function, behind-the-scenes, Huey is doing something like this:
.. code-block:: python
@huey.task()
def add(a, b):
return a + b
result = add(1, 2)
# Is equivalent to:
task = add.s(1, 2)
result = huey.enqueue(task)
Typically, one will only use the :py:meth:`TaskWrapper.s` helper when
creating task execution pipelines.
For example:
.. code-block:: python
add_task = add.s(1, 2) # Represent task invocation.
pipeline = (add_task
.then(add, 3) # Call add() with previous result and 3.
.then(add, 4) # etc...
.then(add, 5))
results = huey.enqueue(pipeline)
# Print results of above pipeline.
print(results.get(blocking=True))
# [3, 6, 10, 15]
.. py:method:: map(it)
:param it: a list, tuple or generic iterable that contains the
arguments for a number of individual task executions.
:returns: a :py:class:`ResultGroup` encapsulating the individual
:py:class:`Result` handlers for the task executions.
.. note::
The iterable should be a list of argument tuples which will be
passed to the task function.
Example:
.. code-block:: python
@huey.task()
def add(a, b):
return a + b
rg = add.map([(i, i * i) for i in range(10)])
# Resolve all results.
rg.get(blocking=True)
# [0, 2, 6, 12, 20, 30, 42, 56, 72, 90]
.. py:method:: unregister()
:returns: boolean indicating whether the task was found and removed
from the registry.
Remove this task from Huey's internal task registry. After
unregistering, the consumer will no longer be able to deserialize or
execute instances of this task. If the task is a periodic task, it will
also be removed from the periodic task schedule.
This is primarily useful when working with dynamically-created periodic
tasks that need to be removed at runtime.
.. py:class:: Task(args=None, kwargs=None, id=None, eta=None, retries=None, retry_delay=None, retry_backoff=None, priority=None, expires=None, timeout=None, on_complete=None, on_error=None)
:param tuple args: arguments for the function call.
:param dict kwargs: keyword arguments for the function call.
:param str id: unique id, defaults to a UUID if not provided.
:param datetime eta: time at which task should be executed.
:param int retries: automatic retry attempts.
:param int retry_delay: seconds to wait before retrying a failed task.
:param retry_backoff: multiplier applied to ``retry_delay`` after each
failed attempt. See :py:meth:`~Huey.task`.
:param int priority: priority assigned to task, higher numbers are
processed first by the consumer when there is a backlog.
:param expires: set expiration time for task. If the task is not run
before ``expires``, it will be discarded. The ``expires`` parameter
can be either an integer (seconds), a timedelta, or a datetime. For
relative expiration values, the expire time will be resolved when
the task is enqueued.
:param int timeout: set execution timeout for task. Implementation
used by the consumer is specific to the worker type and is NOT
available for threads. Processes use SIGALRM, greenlet uses
``gevent.Timeout``. When using with threads it is necessary to use
cooperative timeout checking. See :ref:`cooperative-timeout` for
example.
:param Task on_complete: Task to execute upon completion of this task.
:param Task on_error: Task to execute upon failure / error.
The ``Task`` class represents the execution of a function. Instances of the
task are serialized and enqueued for execution by the consumer, which
deserializes and executes the task function.
.. note::
You should not need to create instances of :py:class:`Task` directly,
but instead use either the :py:meth:`Huey.task` decorator or
the :py:meth:`TaskWrapper.s` method.
Here's a refresher on how tasks work:
.. code-block:: python
@huey.task()
def add(a, b):
return a + b
ret = add(1, 2)
print(ret.get(blocking=True)) # "3".
# The above two lines are equivalent to:
task_instance = add.s(1, 2) # Create a Task instance.
ret = huey.enqueue(task_instance) # Enqueue the queue task.
print(ret.get(blocking=True)) # "3".
.. py:attribute:: is_timed_out
If task was configured with a ``timeout``, returns True if the task has
exceeded the timeout. The task timer starts when the task begins to
execute (after any pre-execute hooks are fired).
This is a property and the return value is calculated dynamically.
.. py:attribute:: time_remaining
If task was configured with a ``timeout``, returns the amount of time
remaining. For tasks that do not specify a timeout, this property
returns ``float('inf')`` so it is always safe to compare.
This is a property and the return value is calculated dynamically.
.. py:method:: check_timeout()
If task was configured with a ``timeout``, this method provides a
single hook for cooperatively checking whether the task has exceeded
its available time, and if so, raises a :class:`TaskTimeout`. This
exception ensures that the ``SIGNAL_TIMEOUT`` fires for the task.
This method is the only mechanism for enforcing a timeout when using
``thread`` workers.
See :ref:`cooperative-timeout` for discussion.
.. py:method:: then(task, *args, **kwargs)
:param TaskWrapper task: A ``task()``-decorated function.
:param args: Arguments to pass to the task.
:param kwargs: Keyword arguments to pass to the task.
:returns: The parent task.
The :py:meth:`~Task.then` method is used to create task pipelines. A
pipeline is a lot like a unix pipe, such that the return value from the
parent task is then passed (along with any parameters specified by
``args`` and ``kwargs``) to the child task.
Here's an example of chaining some addition operations:
.. code-block:: python
add_task = add.s(1, 2) # Represent task invocation.
pipeline = (add_task
.then(add, 3) # Call add() with previous result and 3.
.then(add, 4) # etc...
.then(add, 5))
result_group = huey.enqueue(pipeline)
print(result_group.get(blocking=True))
# [3, 6, 10, 15]
If the value returned by the parent function is a ``tuple``, then the
tuple will be used to update the ``*args`` for the child function.
Likewise, if the parent function returns a ``dict``, then the dict will
be used to update the ``**kwargs`` for the child function.
Example of chaining fibonacci calculations:
.. code-block:: python
@huey.task()
def fib(a, b=1):
a, b = a + b, a
return (a, b) # returns tuple, which is passed as *args
pipe = (fib.s(1)
.then(fib)
.then(fib))
result_group = huey.enqueue(pipe)
print(result_group.get(blocking=True))
# [(2, 1), (3, 2), (5, 3)]
.. py:method:: error(task, *args, **kwargs)
:param TaskWrapper task: A ``task()``-decorated function.
:param args: Arguments to pass to the task.
:param kwargs: Keyword arguments to pass to the task.
:returns: The parent task.
The :py:meth:`~Task.error` method is similar to the
:py:meth:`~Task.then` method, which is used to construct a task
pipeline, except the ``error()`` task will only be called in the event
of an unhandled exception in the parent task.
.. py:function:: crontab(minute='*', hour='*', day='*', month='*', day_of_week='*'[, strict=False])
Convert a "crontab"-style set of parameters into a test function that will
return ``True`` when a given ``datetime`` matches the parameters set forth in
the crontab.
The argument order matches the standard Linux crontab format: minute, hour,
day, month, day-of-week.
Day-of-week uses 0=Sunday and 6=Saturday.
Acceptable inputs:
- `*` = every distinct value
- `*/n` = run every "n" times, i.e. hours=`*/4` == 0, 4, 8, 12, 16, 20
- `m-n` = run every time m..n
- `m,n` = run on m and n
:param bool strict: cause crontab to raise a ``ValueError`` if an input
does not match a supported input format.
:rtype: a test function that takes a ``datetime`` and returns a boolean
.. note::
It is currently not possible to run periodic tasks with an interval
less than once per minute. If you need to run tasks more frequently,
you can create a periodic task that runs once per minute, and from that
task, schedule any number of sub-tasks to run after the desired delays.
.. py:function:: crontab.hourly()
Convenience function for a task that should run every hour, on the hour.
.. py:function:: crontab.daily()
Convenience function for a task that should run daily, at midnight.
.. py:class:: TaskLock(huey, name)
This class should not be instantiated directly, but is instead returned by
:py:meth:`Huey.lock_task`. This object implements a context-manager or
decorator which can be used to ensure only one instance of the wrapped task
is executed at a given time.
If the consumer executes a task and encounters the
:py:class:`TaskLockedException`, then the task will not be executed, an
error will be logged by the consumer, and a ``SIGNAL_LOCKED`` signal will
be emitted. If the task is configured with retries, it will be retried
normally (the lock is released, so the retry has a chance to acquire it).
See :py:meth:`Huey.lock_task` for example usage.
.. py:method:: clear()
Helper method to manually clear the lock. This method is provided to
allow the lock to be flushed in the event that the consumer process was
killed while executing a task holding the lock.
Alternatively, at start-up time you can execute the consumer with the
``-f`` method which will flush all locks before beginning to execute
tasks.
.. py:method:: acquire()
Acquire the lock, raising a :py:class:`TaskLockedException` if the lock
could not be acquired. Otherwise, returns ``True``.
.. py:method:: release()
Identical to :py:meth:`~TaskLock.clear`.
.. py:method:: locked()
:returns: boolean whether lock is currently being held.
.. py:class:: RateLimit(huey, name, limit, per, retry=True)
This class should not be instantiated directly, but is instead returned by
:py:meth:`Huey.rate_limit`. This object implements a context-manager or
decorator which can be used to enforce a simple fixed-window rate-limit of
the wrapped block.
When a task fails due to being rate-limited, it will be automatically
rescheduled to run at the start of the next window when ``retry=True`` or
the task itself has one or more ``retries``.
Interaction between task retries and the ``retry`` flag on the rate-limit:
* ``retry=True``, task has no retries: task will be retried at the start of
the next window (or ``task.retry_delay`` if specified).
* ``retry=True``, task has 1 or more retries: same as above, except task's
own retry count is *preserved*.
* ``retry=False``, task has no retries: error is the final state, task is
not rescheduled.
* ``retry=False``, task has 1 or more retries: task retries decrement
normally and the task's ``retry_delay`` is honored.
Rate-limited tasks emit a ``SIGNAL_RATE_LIMITED``.
:param str name: Name to use for the rate-limiter.
:param int limit: Maximum invocations in the given window.
:param int per: Size of rate-limiting window in seconds.
:param bool retry: Automatically reschedule rate-limited tasks to run
at beginning of next rate-limit window.
:returns: :py:class:`RateLimit` instance, which can be used as a
decorator or context-manager.
.. py:method:: reset()
Reset the rate-limiter, clearing the current window and counter.
The next call to a rate-limited task will start a fresh window.
.. py:method:: current_usage()
:returns: the number of invocations in the current rate-limit window.
Returns ``0`` if no invocations have occurred in this window.
.. py:method:: acquire()
Check whether the rate-limit has been exceeded. If so, raises
:py:class:`RateLimitExceeded`. Otherwise, increments the counter and
returns normally. This is called automatically when the rate-limiter
is used as a decorator or context-manager.
.. py:class:: group(tasks)
A ``group`` is a lightweight wrapper around any number or types of tasks
that may be executed in parallel. When a group is enqueued, Huey returns a
:py:class:`ResultGroup` for reading the results.
:param list tasks: any number of :py:class:`Task` instances.
Example:
.. code-block:: python
@huey.task()
def health_check(service):
return service.name, perform_health_check(service)
g = group([
health_check.s(api_service),
health_check.s(cache_service),
health_check.s(db_service),
health_check.s(proxy_service),
])
result_group = huey.enqueue(g)
for name, is_healthy in result_group(blocking=True):
print('%s healthy? %s' % (name, is_healthy))
.. py:method:: then(task, *args, **kwargs)
:param task: A ``task()``-decorated function (:py:class:`TaskWrapper`),
or a :py:class:`Task` instance to run with the group results.
:param args: Arguments to pass to the callback task.
:param kwargs: Keyword arguments to pass to the callback task.
:returns: A :py:class:`chord` instance.
Convert a :py:class:`group` into a :py:class:`chord` by attaching a
callback. When all members have finished executing, the callback will
be enqueued with the member results (as a list, in order).
.. code-block:: python
result = huey.enqueue(
group([fetch.s(u) for u in urls])
.then(aggregate))
.. py:method:: error(task, *args, **kwargs)
:param task: A ``task()``-decorated function (:py:class:`TaskWrapper`),
or a :py:class:`Task` instance to run with the group results.
:param args: Arguments to pass to the error handler.
:param kwargs: Keyword arguments to pass to the error handler.
:returns: the group instance.
Attach an error handler to every member of the group. If any member
raises an exception, the error handler will be enqueued with the
exception as an argument. Supports chaining:
.. code-block:: python
result = huey.enqueue(
group([fetch.s(u) for u in urls])
.error(on_fetch_error)
.then(aggregate))
.. py:class:: chord(tasks, callback)
A :py:class:`chord` consists of a group of tasks that execute in parallel
and a callback task. When all sub-tasks have completed (or permanently
failed), their results are passed in order as a list to the callback task.
:param list tasks: a list of :py:class:`Task` instances (including
pipelines built with :py:meth:`Task.then`), :py:class:`chord`
instances (for nesting), in any combination.
:param callback: a ``task()``-decorated function (:py:class:`TaskWrapper`),
or a :py:class:`Task` instance. The callback receives the list of
sub-task results as its first argument.
.. code-block:: python
@huey.task()
def fetch(url):
return requests.get(url).json()
@huey.task()
def aggregate(results):
return {r['id']: r for r in results}
c = chord(
[fetch.s(u) for u in urls],
aggregate.s())
result = huey.enqueue(c)
# result is a ChordResult.
data = result(blocking=True) # Blocks until callback finishes.
Enqueueing a chord returns a :py:class:`ChordResult`.
.. note::
The callback always receives the member results as a list, even if
there is only one member. Results are ordered by the position of the
member in the original list, not by the order in which members
finished executing.
**Sub-task errors:**
Chords are designed to always complete. If a sub-task fails and exhausts
all its retries, the exception is used as the task result. Tasks with
retries will retry normally, only after retries are exhausted is the
exception used.
The callback may therefore receive a mix of normal return values and
``Exception`` objects. See :ref:`groups-and-chords` in the guide for
patterns on handling mixed results.
.. note::
Revoking a chord member prevents it from running and will stall the
chord. If you need to cancel a chord, revoke the callback task instead.
**Pipeline members:**
When a chord member is a pipeline (e.g. ``fetch.s(u).then(parse)``), Huey
walks the ``on_complete`` chain and uses the end of the pipeline for the
sub-task. This ensures the chord callback is not triggered until all
pipeline members have fully finished.
Error callbacks placed on individual tasks within a pipeline member
operate independently of the chord.
.. code-block:: python
c = chord(
[download.s(u).then(parse).then(validate) for u in urls],
aggregate.s())
.. py:method:: then(task, *args, **kwargs)
:param task: A ``task()``-decorated function (:py:class:`TaskWrapper`),
or a :py:class:`Task` instance.
:param args: Arguments to pass to the task.
:param kwargs: Keyword arguments to pass to the task.
:returns: the :py:class:`chord` instance.
Chain an additional task after the chord's callback. The callback's
return value is passed to the chained task, exactly like
:py:meth:`Task.then`. Multiple calls to ``then()`` extend the chain.
When tasks are chained to the chord callback, the :py:class:`ChordResult`
returned by :py:meth:`Huey.enqueue` will have a non-``None``
``pipeline_results`` attribute containing a :py:class:`ResultGroup`
for the full callback pipeline.
.. code-block:: python
result = huey.enqueue(
chord([fetch.s(u) for u in urls], aggregate.s())
.then(generate_report)
.then(send_email))
# result.pipeline_results is a ResultGroup with three entries:
# [aggregate result, generate_report result, send_email result]
Delegates to the callback's :py:meth:`Task.then`.
.. py:method:: error(task, *args, **kwargs)
:param task: A ``task()``-decorated function (:py:class:`TaskWrapper`),
or a :py:class:`Task` instance.
:param args: Arguments to pass to the error handler.
:param kwargs: Keyword arguments to pass to the error handler.
:returns: the :py:class:`chord` instance.
Attach an error handler to the chord's callback. If the callback
raises an exception, this task will be enqueued with the exception
as an argument.
.. code-block:: python
result = huey.enqueue(
chord([fetch.s(u) for u in urls], aggregate.s())
.error(alert_admin))
This handles errors in the **callback**, not in the sub-tasks. To handle
errors in individual members, use :py:meth:`Task.error` on each member
or :py:meth:`group.error` before constructing the chord.
Delegates to the callback's :py:meth:`Task.error`.
**Nested chords:**
Chord members can themselves be chords. The inner chord's callback
pipeline serves as the synchronization point for the outer chord.
.. code-block:: python
inner1 = chord([a.s(), b.s()], merge.s())
inner2 = chord([c.s(), d.s()], merge.s())
result = huey.enqueue(chord([inner1, inner2], final.s()))
When the inner chords complete and their callbacks finish, the outer
chord's callback (``final``) receives the two inner callback results as a
list. See :ref:`groups-and-chords` in the guide for a detailed example.
.. py:class:: ChordResult(results, callback_result, pipeline=None)
Returned by :py:meth:`Huey.enqueue` when enqueueing a :py:class:`chord`.
Provides access to the individual member results, the callback result,
and the results of any tasks chained after the callback.
.. py:attribute:: results
A :py:class:`ResultGroup` containing the :py:class:`Result` handles
for each chord member.
.. py:attribute:: callback
A :py:class:`Result` handle for the callback task.
.. py:attribute:: pipeline_results
A :py:class:`ResultGroup` containing :py:class:`Result` handles for
the callback and every task chained after it via
:py:meth:`chord.then`. The first entry is the callback itself, followed
by each chained task in order.
``None`` if no tasks were chained after the callback.
.. code-block:: python
result = huey.enqueue(
chord([a.s(), b.s()], combine.s())
.then(report)
.then(archive))
result.pipeline_results[0]() # combine result
result.pipeline_results[1]() # report result
result.pipeline_results[2]() # archive result
.. py:method:: get(*args, **kwargs)
Shortcut for ``self.callback.get(*args, **kwargs)``. Blocks until the
callback finishes and returns its result.
.. py:method:: __call__(*args, **kwargs)
Identical to :py:meth:`~ChordResult.get`.
.. py:method:: reset()
Reset the callback result and allow re-fetching a new result for the
given task (i.e. after a task error and subsequent retry).
Result
------
.. py:class:: Result(huey, task)
Although you will probably never instantiate an ``Result`` object yourself,
they are returned whenever you execute a task-decorated function, or
schedule a task for execution. The ``Result`` object talks to the result
store and is responsible for fetching results from tasks.
Once the consumer finishes executing a task, the return value is placed in
the result store, allowing the original caller to retrieve it.
Getting results from tasks is straightforward:
.. code-block:: python
>>> @huey.task()
... def add(a, b):
... return a + b
...
>>> res = add(1, 2)
>>> res # what is "res" ?
>>> res() # Fetch the result of this task.
3
What happens when data isn't available yet? Let's assume the next call
takes about a minute to calculate::
>>> res = add(100, 200) # Imagine this is very slow.
>>> res.get() # Data is not ready, so None is returned.
>>> res() is None # We can omit ".get", it works the same way.
True
>>> res(blocking=True, timeout=5) # Block for up to 5 seconds
Traceback (most recent call last):
File "", line 1, in
File "/home/charles/tmp/huey/src/huey/huey/queue.py", line 46, in get
raise ResultTimeout
huey.exceptions.ResultTimeout
>>> res(blocking=True) # No timeout, will block until it gets data.
300
If the task failed with an exception, then a :py:class:`TaskException` will
be raised when reading the result value::
>>> @huey.task()
... def fails():
... raise Exception('I failed')
>>> res = fails()
>>> res() # raises a TaskException!
Traceback (most recent call last):
File "", line 1, in
File "/home/charles/tmp/huey/src/huey/huey/api.py", line 684, in get
raise TaskException(result.metadata)
huey.exceptions.TaskException: Exception('I failed',)
.. py:attribute:: id
Returns the unique id of the corresponding task.
.. py:method:: is_ready()
:returns: whether the task result value is available.
:rtype: bool
.. py:method:: get(blocking=False, timeout=None, backoff=1.15, max_delay=1.0, revoke_on_timeout=False, preserve=False)
:param bool blocking: whether to block while waiting for task result
:param timeout: number of seconds to block (if ``blocking=True``)
:param backoff: amount to backoff delay each iteration of loop
:param max_delay: maximum amount of time to wait between iterations when
attempting to fetch result.
:param bool revoke_on_timeout: if a timeout occurs, revoke the task,
thereby preventing it from running if it is has not started yet.
:param bool preserve: when set to ``True``, this parameter ensures that
the task result will be preserved after having been successfully
retrieved. Ordinarily, Huey will discard results after they have
been read, to prevent the result store from growing without bounds.
:raises: ResultTimeout if blocking and timeout specified without result
becoming ready yet.
Attempt to retrieve the return value of a task. By default,
:py:meth:`~Result.get` will check for the value, returning
``None`` if it is not ready yet. If you want to wait for a value, you
can specify ``blocking=True``. This will loop, backing off up to the
provided ``max_delay``, until the value is ready or the ``timeout`` is
reached. If the ``timeout`` is reached before the result is ready, a
:py:class:`ResultTimeout` exception will be raised.
.. note:: Instead of calling ``.get()``, you can call the
:py:class:`Result` object directly. Both methods accept the same
arguments.
.. py:method:: __call__(**kwargs)
Identical to the :py:meth:`~Result.get` method, provided as a shortcut.
.. py:method:: revoke(revoke_once=True)
:param bool revoke_once: revoke only once.
Revoke the given task. Unless it is in the process of executing, the
task will be discarded without being executed.
.. code-block:: python
one_hour = datetime.datetime.now() + datetime.timedelta(hours=1)
# Run this command in an hour
res = add.schedule((1, 2), eta=one_hour)
# I changed my mind, do not run it after all.
res.revoke()
.. py:method:: restore()
Restore the given task instance. Unless the task instance has already
been dequeued and discarded, it will be restored and run as scheduled.
.. warning::
If the task class itself has been revoked, via a call to
:py:meth:`TaskWrapper.revoke`, then this method has no effect.
.. py:method:: is_revoked()
Return a boolean value indicating whether this particular task instance
**or** the task class itself has been revoked.
.. seealso:: :py:meth:`TaskWrapper.is_revoked`.
.. py:method:: reschedule(eta=None, delay=None, expires=None, priority=None, preserve_pipeline=True)
:param datetime eta: execute function at the given time.
:param int delay: execute function after specified delay in seconds.
:param expires: set expiration time for task. If not provided, then the
task's original expire time (if any) will be used.
:param priority: override the task's priority. If not provided, the
original priority is preserved.
:param bool preserve_pipeline: when ``True`` (default), the
rescheduled task retains any ``on_complete`` and ``on_error``
chains from the original task. Set to ``False`` to discard them.
:returns: :py:class:`Result` handle for the new task.
Reschedule the given task. The original task instance will be revoked,
but **no checks are made** to verify that it hasn't already been
executed.
If neither an ``eta`` nor a ``delay`` is specified, the task will be
run as soon as it is received by a worker.
.. py:method:: reset()
Reset the cached result and allow re-fetching a new result for the
given task. This is essential when a task fails and is retried: the
first call to :py:meth:`~Result.get` caches the error locally, and
subsequent calls return the cached error even after the retry succeeds.
Calling ``reset()`` clears the local cache so the next ``get()`` reads
from the result store again.
.. code-block:: python
result = flaky_task()
try:
result.get(blocking=True, timeout=5)
except TaskException:
result.reset() # Clear cached error.
# Now we can read the result from the successful retry:
value = result.get(blocking=True, timeout=30)
.. py:class:: ResultGroup
A ``ResultGroup`` will be returned when you enqueue a task pipeline or if
you use the :py:meth:`TaskWrapper.map` method. It is a simple wrapper
around a number of individual :py:meth:`Result` instances, and provides a
convenience API for fetching the results in bulk.
.. py:method:: get(**kwargs)
Call :py:meth:`~Result.get` on each individual :py:meth:`Result`
instance in the group and returns a list of return values. Any keyword
arguments are passed along.
.. py:method:: as_completed(backoff=1.15, max_delay=1.0)
:param float backoff: factor to increase delay between polls.
:param float max_delay: maximum seconds between polls.
:returns: a generator that yields individual result values as they
become available.
Iterate over results in the order they become ready, rather than the
order they were enqueued. This is useful when you want to begin
processing results as soon as possible without waiting for all tasks
to finish.
.. code-block:: python
result_group = huey.enqueue(group([
fetch.s(url) for url in urls]))
for value in result_group.as_completed():
process(value)
The generator polls each result handle in a round-robin fashion with
exponential backoff, yielding values as they appear.
.. py:method:: __getitem__(idx)
Resolves the result at the given index, blocking until it is ready.
.. py:method:: __iter__()
Iterate over the individual :py:class:`Result` instances in the group.
.. py:method:: __len__()
Return the number of results in the group.
Consumer
--------
.. py:class:: Consumer(huey, workers=1, periodic=True, initial_delay=0.1, backoff=1.15, max_delay=10.0, scheduler_interval=1, worker_type='thread', check_worker_health=True, health_check_interval=10, flush_locks=False, extra_locks=None, max_tasks=None)
:param huey: the :py:class:`Huey` instance whose queue will be consumed.
Coordinates the workers and the scheduler, and installs the shutdown signal
handlers. This is what the ``huey_consumer`` command-line tool runs, and
what :py:meth:`Huey.create_consumer` returns. The keyword arguments
correspond one-to-one with the command-line options documented in
:ref:`consumer-options`.
Instantiate it directly only when embedding the consumer in another process
or writing tests. Ordinarily, run ``huey_consumer`` (or ``manage.py
run_huey`` under Django) instead.
.. py:method:: run()
Start the consumer and block until it is shut down, either by a signal
or by :py:meth:`stop`. This is what ``huey_consumer`` calls.
.. py:method:: start()
Start the worker and scheduler processes and register the signal
handlers, then return immediately. Raises
:py:class:`ConfigurationError` if the :py:class:`Huey` instance has
``immediate`` enabled.
.. py:method:: stop(graceful=False)
:param bool graceful: block until in-flight tasks finish.
Set the stop flag. With ``graceful=False`` (default) the method returns
as soon as the flag is set, interrupting any running tasks. With
``graceful=True`` it blocks until every worker has finished the task it
is currently executing.
Serializer
----------
.. py:class:: Serializer(compression=False, compression_level=6, use_zlib=False)
:param bool compression: use gzip compression
:param int compression_level: 0 for least, 9 for most.
:param bool use_zlib: use zlib for compression instead of gzip.
The Serializer class implements a simple interface that can be extended to
provide your own serialization format. The default implementation uses
``pickle``.
To override, the following methods should be implemented. Compression is
handled transparently elsewhere in the API.
.. py:method:: _serialize(data)
:param data: arbitrary Python object to serialize.
:rtype bytes:
.. py:method:: _deserialize(data)
:param bytes data: serialized data.
:returns: the deserialized object.
.. py:class:: SignedSerializer(secret, salt='huey', **kwargs)
:param str secret: secret key used to generate HMAC signatures.
:param str salt: salt combined with the secret (default ``'huey'``).
:param kwargs: additional keyword arguments passed to the base
:py:class:`Serializer` (e.g. ``compression``, ``use_zlib``).
A subclass of :py:class:`Serializer` that adds an HMAC-SHA1 signature to
every serialized message. When deserializing, the signature is verified and
a ``ValueError`` is raised if the message has been tampered with.
This is useful when the storage backend (e.g. Redis) is shared or
network-exposed and you want to prevent malicious injection of crafted
pickle payloads.
.. code-block:: python
from huey import RedisHuey
from huey.serializer import SignedSerializer
huey = RedisHuey(
'my-app',
serializer=SignedSerializer(secret='my-secret-key'))
.. note::
The signed serializer detects tampering but does **not** encrypt the
data. Task arguments remain visible in the storage backend.
Both the application and the consumer must use the same ``secret`` and
``salt``.
.. _exceptions:
Exceptions
----------
.. py:class:: HueyException
General exception class.
.. py:class:: ConfigurationError
Raised when Huey encounters a configuration problem.
.. py:class:: TaskLockedException
Raised by the consumer when a task lock cannot be acquired.
.. py:class:: ResultTimeout
Raised when attempting to block on a call to :py:meth:`Result.get` (for
instance) and the timeout is exceeded without the result being ready.
.. py:class:: TaskTimeout
Raised when a task exceeds its specified execution time.
.. py:class:: RateLimitExceeded(key, delay, retry=True)
Raised when a rate-limit has been exceeded.
.. py:class:: CancelExecution(retry=None)
Cancel the execution of a task. Can be raised either within a
:py:meth:`~Huey.pre_execute` hook, or within a
:py:meth:`~Huey.task`-decorated function.
When raised from within a :py:meth:`~Huey.pre_execute` hook, this exception
signals to the consumer that the task shall be cancelled and not run.
When raised in the body of a :py:meth:`~Huey.task`-decorated function, the
``retry`` parameter controls whether the task is retried:
* ``retry=None`` (default): the task's normal retry policy applies. If the
task has retries remaining, they are decremented and the task is
re-enqueued. If no retries remain, the task is not retried.
* ``retry=True``: the task will be retried regardless of whether it was
declared with ``retries``. If the task has no retries, Huey sets the
retry count to 1 so at least one retry occurs.
* ``retry=False``: the task will **not** be retried, even if it has
retries remaining.
.. py:class:: RetryTask(msg=None, eta=None, delay=None)
Raised by user code from within a :py:meth:`~Huey.task` function to force a
retry. When this exception is raised, the task will be retried irrespective
of whether it is configured with automatic retries.
If ``delay`` or ``eta`` is specified, then any ``retry_delay`` set on the
task will be overridden and the value specified will be used to determine
when the task will be retried next.
.. code-block:: python
@huey.task()
def fetch_api_data(url):
try:
fh = urlopen(url)
except HTTPError:
# Try again in 60 seconds for an HTTP error (500, etc).
raise RetryTask(delay=60)
...
.. py:class:: TaskException
General exception raised by :py:class:`Result` handles when reading the
result of a task that failed due to an error.
Storage
-------
Huey comes with several built-in storage implementations:
.. py:class:: RedisStorage(name='huey', blocking=True, read_timeout=1, connection_pool=None, url=None, client_name=None, notify_result=False, notify_result_ttl=86400, **connection_params)
:param str name: namespace for storage.
:param bool blocking: Use blocking-pop when reading from the queue (as
opposed to polling). Default is true.
:param read_timeout: Timeout to use when performing a blocking pop, default
is 1 second.
:param connection_pool: a redis-py ``ConnectionPool`` instance.
:param url: url for Redis connection.
:param client_name: name used to identify Redis clients used by Huey.
:param bool notify_result: use a blocking-pop on a result-ready key to
enable low-latency result reading.
:param int notify_result_ttl: TTL for result-ready key to automatically
expire un-awaited results.
Additional keyword arguments will be passed directly to the Redis client
constructor. See the `redis-py documentation `_
for the complete list of arguments supported by the Redis client.
.. py:class:: RedisExpireStorage(name='huey', expire_time=86400, blocking=True, read_timeout=1, connection_pool=None, url=None, client_name=None, notify_result=False, notify_result_ttl=86400, **connection_params)
:param int expire_time: TTL for results of individual tasks.
Subclass of :py:class:`RedisStorage` that implements the result store APIs
using normal Redis keys with a TTL, so that unread results will
automatically be cleaned-up. :py:class:`RedisStorage` uses a *HASH* for the
result store, which has the benefit of keeping the Redis keyspace orderly,
but which comes with the downside that unread task results can build up
over time. This storage implementation trades keyspace sprawl for automatic
clean-up.
.. py:class:: PriorityRedisStorage(name='huey', blocking=True, read_timeout=1, connection_pool=None, url=None, client_name=None, notify_result=False, notify_result_ttl=86400, **connection_params)
:param str name: namespace for storage.
:param bool blocking: Use blocking-zpopmin when reading from the queue (as
opposed to polling). Default is true.
:param read_timeout: Timeout to use when performing a blocking pop, default
is 1 second.
:param connection_pool: a redis-py ``ConnectionPool`` instance.
:param url: url for Redis connection.
:param client_name: name used to identify Redis clients used by Huey.
:param bool notify_result: use a blocking-pop on a result-ready key to
enable low-latency result reading.
:param int notify_result_ttl: TTL for result-ready key to automatically
expire un-awaited results.
Redis storage that uses a different data-structure for the task queue in
order to support task priorities.
Additional keyword arguments will be passed directly to the Redis client
constructor. See the `redis-py documentation `_
for the complete list of arguments supported by the Redis client.
.. warning:: This storage engine requires Redis 5.0 or newer.
.. py:class:: PriorityRedisExpireStorage(name='huey', expire_time=86400, ...)
:param int expire_time: TTL for results of individual tasks.
Combination of :py:class:`PriorityRedisStorage`, which supports task
priorities, and :py:class:`RedisExpireStorage`, which stores task results
as top-level Redis keys in order set a TTL so that unread results are
automatically cleaned-up.
.. py:class:: SqliteStorage(name='huey', filename='huey.db', cache_mb=8, fsync=None, journal_mode='wal', timeout=5, strict_fifo=False, create_tables=True, **kwargs)
:param str name: namespace for storage.
:param str filename: sqlite database filename.
:param int cache_mb: sqlite page-cache size in megabytes.
:param bool fsync: if ``True``, all writes to the Sqlite database will be
synchronized. This provides greater safety from database corruption in
the event of sudden power-loss. If set to ``False`` then let the OS
handle syncing to disk. Uses the library default if unspecified.
:param str journal_mode: sqlite journaling mode to use. Defaults to using
write-ahead logging, which enables readers to coexist with a single
writer.
:param int timeout: busy timeout (in seconds), amount of time to wait to
acquire the write lock when another thread / connection holds it.
:param bool strict_fifo: ensure that the task queue behaves as a strict
FIFO. By default, Sqlite may reuse rowids for deleted tasks, which can
cause tasks to be run in a different order than the order in which they
were enqueued.
:param bool create_tables: create tables if they do not exist.
:param kwargs: Additional keyword arguments passed to the ``sqlite3``
connection constructor.
.. py:class:: CySqliteStorage(name='huey', filename='huey.db', pragmas=None, timeout=5, strict_fifo=False, create_tables=True, **kwargs)
Extends :py:class:`SqliteStorage`, using `cysqlite `_
as the driver instead of the standard library ``sqlite3`` module.
:param str name: namespace for storage.
:param str filename: sqlite database filename.
:param dict pragmas: mapping of pragmas to apply to every connection. If
unset, ``journal_mode`` will default to ``'wal'``, which enables
readers to coexist with a single writer.
:param int timeout: busy timeout (in seconds), amount of time to wait to
acquire the write lock when another thread / connection holds it.
:param bool strict_fifo: ensure that the task queue behaves as a strict
FIFO. By default, Sqlite may reuse rowids for deleted tasks, which can
cause tasks to be run in a different order than the order in which they
were enqueued.
:param bool create_tables: create tables if they do not exist.
:param kwargs: Additional keyword arguments passed to
``cysqlite.connect()``.
Where :py:class:`SqliteStorage` exposes a fixed set of tuning parameters,
this storage takes an open-ended ``pragmas`` dict. The
:py:class:`SqliteStorage` parameters ``cache_mb``, ``fsync`` and
``journal_mode`` are still accepted and are translated into the
corresponding ``cache_size``, ``synchronous`` and ``journal_mode`` pragmas.
.. py:class:: PostgresStorage(name='huey', dsn=None, connection=None, blocking=True, read_timeout=1, table_prefix='huey', create_tables=True, **connection_params)
:param str name: namespace for storage.
:param str dsn: connection string or URL for ``psycopg.connect()``.
:param connection: zero-argument callable returning a new ``psycopg``
connection.
:param bool blocking: use LISTEN/NOTIFY to block on dequeue rather than
polling. Default is true.
:param read_timeout: timeout to use when performing a blocking dequeue,
default is 1 second.
:param str table_prefix: prefix for huey's tables, default ``huey``.
:param bool create_tables: issue ``create table if not exists`` for huey's
tables when the storage is instantiated. Default is true. Set to false
to manage the schema yourself, e.g. via Django migrations or the
``create_huey_tables`` management command.
:param connection_params: additional keyword arguments passed directly to
``psycopg.connect()``.
Requires psycopg 3.2 or newer. Dequeues use ``select ... for update skip locked``,
so any number of consumers can safely share one database.
.. note::
PgBouncer in transaction-pooling mode does not support LISTEN. Either
use session pooling or a direct connection for huey, or specify
``blocking=False`` to use polling instead.
.. note::
When using the greenlet worker model, install psycopg in pure-Python
mode (the C extension bypasses the sockets patched by gevent).
.. py:class:: FileStorage(name, path, levels=2, use_thread_lock=False)
:param str name: (unused by the file storage API)
:param str path: directory path used to store task results. Will be created
if it does not exist.
:param int levels: number of levels in cache-file directory structure to
ensure a given directory does not contain an unmanageable number of
files.
:param bool use_thread_lock: use the standard lib ``threading.Lock``
instead of a lockfile. Note: this should only be enabled when using the
greenlet or thread consumer worker models.
The :py:class:`FileStorage` implements a simple file-system storage layer.
This storage class should not be used in high-throughput, highly-concurrent
environments, as it utilizes exclusive locks around all file-system
operations. This is done to prevent race-conditions when reading from the
file-system.
.. py:class:: MemoryStorage()
In-memory storage engine for use when testing or developing. Designed for
use with :ref:`immediate mode `.
.. py:class:: BlackHoleStorage()
Storage class that discards all data written to it, and thus always appears
to be empty. Intended for testing only.
.. autoclass:: huey.storage.BaseStorage
:members:
---
## File: docs/asyncio.rst
.. _asyncio:
AsyncIO
-------
While Huey does not provide first-class support for a full asyncio pipeline, in
practice one of the most useful locations to be "async"-friendly is when
blocking while waiting for a task result to be ready. When waiting for a task
result, Huey must poll the storage backend to determine if the result is ready
which means lots of opportunity for an asynchronous solution.
To simplify this, Huey provides two helpers for ``await``-ing task
results. For a complete example of wiring Huey into an async web application,
see :ref:`recipe-fastapi`.
.. py:function:: aget_result(result, backoff=1.15, max_delay=1.0, preserve=False)
:param Result result: a result handle returned when calling a task.
:return: task return value.
AsyncIO helper for awaiting the result of a task execution.
Example:
.. code-block:: python
@huey.task()
def sleep(n):
time.sleep(n)
return n
async def main():
# Single task, will finish in ~2 seconds (other coroutines can run
# during this time!).
rh = sleep(2)
result = await aget_result(rh)
# Awaiting multiple results. This will also finish in ~2 seconds.
r1 = sleep(2)
r2 = sleep(2)
r3 = sleep(2)
results = await asyncio.gather(
aget_result(r1),
aget_result(r2),
aget_result(r3))
.. py:function:: aget_result_group(rg, *args, **kwargs)
:param ResultGroup rg: a result-group handle for multiple tasks.
:return: return values for all tasks in the result group.
AsyncIO helper for awaiting the result of multiple task executions.
Example:
.. code-block:: python
@huey.task()
def sleep(n):
time.sleep(n)
return n
async def main():
# Spawn 3 "sleep" tasks, each sleeping for 2 seconds.
rg = sleep.map([2, 2, 2])
# Await the results. This will finish in ~2 seconds while also
# allowing other coroutines to run.
results = await aget_result_group(rg)
---
## File: docs/changes.rst
.. _changes:
Changes in 3.0
==============
The 3.0 release of Huey is API-compatible with the previous versions. The main
change is the addition of a number of new features:
* :py:class:`chord` and :py:class:`group` - :ref:`description `
* :ref:`task timeouts `
* :ref:`rate limiting `
* low-latency result notification (Redis-only) via ``notify_result=True``.
* removed Python 2.x compatibility.
Changes in 2.0
==============
The 2.0 release of Huey is mostly API-compatible with previous versions, but
there are a number of things that have been altered or improved in this
release.
.. warning::
The serialization format for tasks has changed. An attempt has been made to
provide backward compatibility when reading messages enqueued by an older
version of Huey, but this is not guaranteed to work.
Summary
-------
The ``always_eager`` mode has been renamed :ref:`immediate`. As the new name
implies, tasks are run immediately instead of being enqueued. Immediate mode is
designed to be used during testing and development. When immediate mode is
enabled, Huey switches to using in-memory storage by default, so as to avoid
accidental writes to a live storage. Immediate mode improves greatly on
``always_eager`` mode, as it no longer requires special-casing and follows the
same code-paths used when Huey is in live mode. See :ref:`immediate` for more
details.
Previously, the Huey consumer accepted options to run in UTC or local-time.
Various APIs, particularly around scheduling and task revocation, needed to be
compatible with however the consumer was configured, and it could easily get
confusing. As of 2.0, UTC-vs-localtime is specified when instantiating Huey,
and all conversion happens internally, which should make things easier to
think about.
The events APIs have been removed and replaced by a :ref:`signals` system.
Signal handlers are executed synchronously by the worker(s) as they run, which
is easier to use than the events API. That API used a pub/sub approach and
required a dedicated listener thread to make any use of it.
Errors are no longer stored in a separate list. Should a task fail due to an
unhandled exception, the exception will be placed in the result store, and can
be introspected using the task's :py:class:`Result` handle.
Huey now supports :ref:`priority`. To use priorities with Redis, you need to be
running Redis 5.0 or newer, and should use :py:class:`PriorityRedisHuey`. The
original :py:class:`RedisHuey` continues to support older versions of Redis.
:py:class:`SqliteHuey` and the in-memory storage used for dev/testing provide
full support for task priorities.
Details
-------
Changes when initializing :py:class:`Huey`:
* ``result_store`` parameter has been renamed to ``results``.
* ``events`` parameter is removed. Events have been replaced by :ref:`signals`.
* ``store_errors`` parameter is removed. Huey no longer maintains a separate
list of recent errors. Unhandled errors that occur when running a task are
stored in the result store. Also the ``max_errors`` parameter of the Redis
storage engine is removed.
* ``global_registry`` parameter is removed. Tasks are registered to the Huey
instance with which they are decorated, rather than to a global registry.
* ``always_eager`` has been renamed ``immediate``.
New initialization arguments:
* Boolean ``utc`` parameter (defaults to true). This setting is used to control
how Huey interprets datetimes internally. Previously, this logic was spread
across a number of APIs and a consumer flag.
* Serializer parameter accepts an (optional) object implementing the
:py:class:`Serializer` interface. Defaults to using ``pickle``.
* Accepts option to use gzip ``compression`` when serializing data.
Other changes to :py:class:`Huey`:
* Immediate mode can be enabled or disabled at runtime by setting the
:py:attr:`~Huey.immediate` property.
* Event emitter has been replaced by :ref:`signals`, so all event-related APIs
have been removed.
* Special classes of exceptions for the various storage operations have been
removed. For more information see :ref:`exceptions`.
* The ``Huey.errors()`` method is gone. Errors are no longer tracked
separately.
Changes to the :py:meth:`~Huey.task` and :py:meth:`~Huey.periodic_task`
decorators:
* Previously these decorators accepted two optional keyword arguments,
``retries_as_argument`` and ``include_task``. Since the remaining retries are
stored as an attribute on the task itself, the first is redundant. In 2.0
these are replaced by a new keyword argument ``context``, which, if ``True``,
will pass the task instance to the decorated function as a keyword argument.
* Enqueueing a task pipeline will now return a :py:class:`ResultGroup` instead
of a list of individual :py:class:`Result` instances.
Changes to the :py:class:`Result` handle (previous called
``TaskResultWrapper``):
* The ``task_id`` property is renamed to ``id``.
* Task instances that are revoked via :py:meth:`Result.revoke` will default to
using ``revoke_once=True``.
* The :py:meth:`~Result.reschedule` method no longer requires a delay or eta.
Leaving both empty will reschedule the task immediately.
Changes to :py:func:`crontab`:
* The order of arguments has been changed to match the order used on linux
crontab. The order is now minute, hour, day, month, day of week.
Miscellaneous:
* Huey no longer uses a global registry for task functions. Task functions are
only visible to the huey instance they are decorated by.
* ``RedisHuey`` defaults to using a blocking pop on the queue, which should
improve latency and reduce chatter. To go back to the old polling default,
specify ``blocking=False`` when creating your huey instance.
* ``SqliteHuey`` no longer has any third-party dependencies and has been moved
into the main ``huey`` module.
* The :py:class:`MiniHuey` contrib module has been renamed to
``huey.contrib.mini``.
* The ``SimpleStorage`` contrib module has been removed.
Django-specific:
* The ``backend_class`` setting has been renamed to ``huey_class`` (used to
specify import-path to Huey implementation, e.g. ``huey.RedisHuey``).
---
## File: docs/consumer.rst
.. _consuming-tasks:
Consuming Tasks
===============
To run the consumer, point it at the "import path" to your application's
:py:class:`Huey` instance. For example, here is how I run it on my blog:
.. code-block:: shell
huey_consumer blog.main.huey --logfile=../logs/huey.log
The "import path" is the dotted-path you would use to import the "huey" object
in the interactive interpreter:
.. code-block:: pycon
>>> from blog.main import huey
You may run into trouble though when "blog" is not on your python-path. To
work around this:
1. Manually specify your pythonpath: ``PYTHONPATH=/some/dir/:$PYTHONPATH huey_consumer blog.main.huey``.
2. Run the consumer from the directory your config module is in. I use
supervisord to manage my huey process, so I set the ``directory`` to the
root of my site.
3. Create a wrapper and hack ``sys.path``.
.. warning::
If you plan to use `supervisord `_ to manage your
consumer process, be sure that you are running the consumer directly and
without any intermediary shell scripts. Shell script wrappers interfere
with supervisor's ability to terminate and restart the consumer Python
process. For discussion see `GitHub issue 88 `_.
.. _consumer-options:
Options for the consumer
------------------------
The following table lists the options available for the consumer as well as
their default values.
``-l``, ``--logfile``
Path to file used for logging. The logfile grows indefinitely, so you may
wish to configure a tool like ``logrotate``.
Alternatively, you can attach your own handler to ``huey.consumer``.
The default loglevel is ``INFO``.
``-v``, ``--verbose``
Verbose logging (loglevel=DEBUG). If no logfile is specified and
verbose is set, then the consumer will log to the console.
**Note:** due to conflicts, when using Django this option is renamed to
use ``-V``, ``--huey-verbose``.
``-q``, ``--quiet``
Minimal logging, only errors and their tracebacks will be logged.
``-S``, ``--simple``
Use a simple log format consisting only of the time H:M:S and log message.
``-w``, ``--workers``
Number of worker threads/processes/greenlets. The default is ``1``, but
most applications will want at least 2, so a slow task cannot tie up the
only worker. For CPU-bound workloads, scale toward the number of CPU cores.
With the ``greenlet`` worker type you can run hundreds of workers, but when
using Redis, size the connection pool to match.
``-k``, ``--worker-type``
Choose the worker type, ``thread``, ``process`` or ``greenlet``. The
default is ``thread``.
Depending on your workload, one worker type may perform better than the
others:
* CPU heavy loads: use "process". Python's global interpreter lock prevents
multiple threads from running simultaneously, so to leverage multiple CPU
cores (and reduce thread contention) run each worker as a separate
process.
* IO heavy loads: use "greenlet". For example, tasks that crawl websites or
which spend a lot of time waiting to read/write to a socket, will get a
huge boost from using the greenlet worker model. Because greenlets are so
cheap in terms of memory, you can easily run a large number of workers.
Note that all code that does **not** consist in waiting for a socket will
be blocking and cannot be pre-empted. Understand the tradeoffs before
jumping to use greenlets. When using with Redis, ensure that your
connection pool is large enough to provide connections for each greenlet.
* Anything else: use "thread". You get the benefits of pre-emptive
multi-tasking without the overhead of multiple processes. A safe choice
and the default.
See the :ref:`worker-types` section for additional information.
``-n``, ``--no-periodic``
Indicate that this consumer process should *not* enqueue periodic tasks.
When running multiple consumers against one queue, all but one should use
this so periodic tasks are enqueued once rather than once per consumer.
``-d``, ``--delay``
When using a "polling"-type queue backend, this is the number of seconds to
wait when polling the backend. Default is 0.1 seconds. For example, when
the consumer starts up it will begin polling every 0.1 seconds. If no tasks
are found in the queue, it will multiply the current delay (0.1) by the
backoff parameter. When a task is received, the polling interval will reset
back to this value.
``-m``, ``--max-delay``
The maximum amount of time to wait between polling, if using weighted
backoff. Default is 10 seconds. If your huey consumer doesn't see a lot of
action, you can increase this number to reduce CPU usage.
``-b``, ``--backoff``
The amount to back-off when polling for results. Must be greater than
one. Default is 1.15. This parameter controls the rate at which the
interval increases after successive attempts return no tasks. Here is how
the defaults, 0.1 initial and 1.15 backoff, look:
.. image:: https://media.charlesleifer.com/blog/photos/p1472257818.22.png
``-c``, ``--health-check-interval``
How often huey checks the status of the workers, restarting any that died.
Default is 10 seconds.
``-C``, ``--disable-health-check``
Disable the worker health checks. Leaving them enabled is cheap and lets
the consumer restart workers that die, so most deployments should keep the
default.
``-f``, ``--flush-locks``
Flush all locks when starting the consumer. This may be useful if the
consumer was killed abruptly while executing a locked task.
``-L``, ``--extra-locks``
Additional lock-names to flush when starting the consumer, separated by
comma. This is useful if you have locks within context-managers that may
not be discovered during consumer startup, but you wish to ensure they are
cleared. Implies ``--flush-locks``.
``-M``, ``--max-tasks``
Restart a worker after it has executed the given number of tasks. This
option requires that the worker health check be enabled. If it is not, an
error will be raised.
``-s``, ``--scheduler-interval``
The frequency with which the scheduler should run. By default this will run
every second, but you can increase the interval to as much as 60 seconds.
The value must divide evenly into 60.
Examples
^^^^^^^^
Running the consumer with 8 threads and a logfile for errors:
.. code-block:: shell
huey_consumer my.app.huey -l /var/log/app.huey.log -w 8 -q
Using multi-processing to run 4 worker processes.
.. code-block:: shell
huey_consumer my.app.huey -w 4 -k process
Running single-threaded with periodic task support disabled. Additionally,
logging records are written to stdout.
.. code-block:: shell
huey_consumer my.app.huey -v -n
Using greenlets to run 50 workers, with no health checking and a scheduler
granularity of 60 seconds.
.. code-block:: shell
huey_consumer my.app.huey -w 50 -k greenlet -C -s 60
.. _worker-types:
Worker types
------------
The consumer consists of a main process, a scheduler, and one or more workers.
These individual components all run concurrently, and Huey supports three
different mechanisms to achieve this concurrency.
* *thread*, the default, uses OS threads. Due to Python's global interpreter
lock, only one thread can be running at a time, but this is actually less of
a limitation than it might sound. The Python runtime can intelligently switch
the running thread when an I/O occurs or when a thread is idle. If the worker
is CPU-bound, the runtime will pre-emptively switch threads after a given
number of operations, ensuring each thread gets a chance to make progress.
Threads provide a good balance of performance and memory efficiency.
* *process* runs the scheduler and worker(s) in their own process. The main
benefit over threads is the absence of the global interpreter lock, which
allows CPU-bound workers to execute in parallel. Since each process maintains
its own copy of the code in memory, it is likely that processes will require
more memory than threads or greenlets. Processes are a good choice for tasks
that perform CPU-intensive work.
* *greenlet* runs the scheduler and worker(s) in greenlets. Requires `gevent `_,
a cooperative multi-tasking library. When a task performs an operation that
would be blocking (read or write on a socket), the file descriptor is added
to an event loop managed by gevent, and the scheduler will switch tasks.
Since gevent uses cooperative multi-tasking, a task that is CPU-bound will
not yield control to the gevent scheduler, limiting concurrency. For this
reason, gevent is a good choice for tasks that perform lots of socket I/O,
but may give worse performance for tasks that are CPU-bound (e.g., parsing
large files, manipulating images, generating reports, etc). Understand the
tradeoff thoroughly before using this worker type. When using Redis, ensure
that your connection pool is large enough for each greenlet to have its own
connection.
When in doubt, the default setting (``thread``) is a safe choice.
.. warning::
Multiprocess support is not available for Windows. The only process start
method available on Windows is "spawn", which has the downside of requiring
the Huey state to be pickled. Huey uses (and creates) many objects which
cannot be pickled. More information here: `multiprocessing documentation `_.
Using gevent
^^^^^^^^^^^^
Gevent works by monkey-patching various Python modules, such as ``socket``,
``ssl``, ``time``, etc. In order for your application to be able to switch
tasks reliably, you should apply the monkey-patch at the very beginning of
your code, before anything else gets loaded.
Suppose we have defined an entrypoint for our application named
``main.py``, which imports our :py:class:`Huey` instance, our tasks, and
the other essential parts of our application (the WSGI app, database
connection, etc).
We would place the monkey-patch at the top of ``main.py``, before all the
other imports:
.. code-block:: python
# main.py
from gevent import monkey; monkey.patch_all() # Apply monkey-patch.
from .app import wsgi_app # Import our WSGI app.
from .db import database # Database connection.
from .queue import huey # Huey instance for our app.
from .tasks import * # Import all tasks, so they are discoverable.
To run the consumer:
.. code-block:: shell
huey_consumer main.huey -k greenlet -w 16
You should have a good understanding of how gevent works, its strengths and
limitations, before using the greenlet worker type.
.. _consumer-shutdown:
Consumer shutdown
-----------------
The huey consumer supports graceful shutdown via ``SIGINT``. When the consumer
process receives ``SIGINT``, workers are allowed to finish up whatever task
they are currently executing before the process exits.
Alternatively, you can shutdown the consumer using ``SIGTERM`` and any running
tasks will be interrupted, ensuring the process exits quickly.
Huey does not guarantee at-least-once delivery of messages, and does not do
acknowledgement of completed tasks. This means that if you terminate the
consumer **without** letting it finish any currently-executing tasks, those
tasks will be lost. To be alerted when this occurs, you can use Huey's
:ref:`signals` (specifically ``signals.SIGNAL_INTERRUPTED``). The consumer
will emit this for tasks that are interrupted during execution.
.. _consumer-deployments:
Deployments
^^^^^^^^^^^
When deploying new code, your best bet is to gracefully shutdown the Huey
consumer using ``SIGINT``, letting all running tasks finish, before starting a
new consumer process using the new code.
If you have long-running tasks, an alternative option is to configure
your new code to use a separate storage namespace. On Redis this is as simple
as specifying a new ``name`` for your ``RedisHuey()`` instance. Then you can
start the new code and new consumer, and they will operate independently of the
previously-running consumer. When all tasks are done, you can gracefully
shutdown the old consumer.
It is always a good idea to implement a Huey ``signals.SIGNAL_INTERRUPTED``
handler (:ref:`signals`), even if all it does is log an exception about the
interrupted task.
.. _consumer-restart:
Consumer restart
----------------
To cleanly restart the consumer, including all workers, send the ``SIGHUP``
signal. When the consumer receives the hang-up signal, any tasks being executed
will be allowed to finish before the restart occurs.
.. _process-supervisors:
supervisord and systemd
-----------------------
Huey plays nicely with both `supervisord `_,
`systemd `_ and presumably any other process supervisor.
For complete deployment examples (Docker, Docker Compose, PaaS
configurations) and a production checklist, see :ref:`deployment`.
.. warning::
Both supervisord and systemd stop processes with ``SIGTERM`` by default,
which huey treats as "stop immediately, interrupting any running tasks".
Huey's graceful-shutdown signal is ``SIGINT``, so be sure to configure
the stop signal as shown below. Otherwise tasks will be interrupted on
every deploy, regardless of how long the supervisor is told to wait.
Barebones supervisor config using 4 worker threads:
.. code-block:: ini
[program:my_huey]
directory=/path/to/project/
command=/path/to/venv/bin/huey_consumer my_app.huey -w 4
user=someuser
autostart=true
autorestart=true
stdout_logfile=/var/log/huey.log
stderr_logfile=/var/log/huey.err
; Note: supervisor does not perform shell expansion, so a literal
; "$PYTHONPATH" cannot be used here.
environment=PYTHONPATH="/path/to/project"
; Huey shuts down gracefully on SIGINT, allowing workers to finish their
; current task. Supervisor's default stopsignal is TERM, which huey
; treats as "stop immediately", so be sure to specify INT here.
stopsignal=INT
; How long to wait for in-flight tasks to finish before escalating to
; SIGKILL. Increase this if you have long-running tasks.
stopwaitsecs=30
Barebones systemd config using 4 worker threads:
.. code-block:: ini
[Unit]
Description=My Huey
After=network.target
[Service]
User=someuser
Group=somegroup
WorkingDirectory=/path/to/project/
ExecStart=/path/to/venv/bin/huey_consumer my_app.huey -w 4
Restart=always
# Huey shuts down gracefully on SIGINT, allowing workers to finish their
# current task. systemd's default KillSignal is SIGTERM, which huey
# treats as "stop immediately", so override it here.
KillSignal=SIGINT
# How long to wait for in-flight tasks to finish before escalating to
# SIGKILL. Increase this if you have long-running tasks.
TimeoutStopSec=60
[Install]
WantedBy=multi-user.target
.. note::
Django users may replace ``huey_consumer`` with the appropriate path to
``manage.py run_huey``.
.. _multiple-consumers:
Multiple Consumers
------------------
Huey is typically run on a single server, with the number of workers scaled-up
according to your applications workload. However, it is also possible to run
multiple Huey consumers across multiple servers. When running multiple
consumers, it is crucial that **only one consumer** be configured to enqueue
periodic tasks.
By default the consumer will enqueue periodic tasks for execution whenever they
are ready to be run. When multiple consumers are used, it is therefore
necessary to specify the ``-n`` or ``--no-periodic`` option for all consumers
except one.
For example:
* Server A (main): ``huey_consumer myapp.huey -w 8 -k process``
* Server B: ``huey_consumer myapp.huey -w 8 -k process --no-periodic``
* Server C: ``huey_consumer myapp.huey -w 8 -k process --no-periodic``
Since each Huey consumer must be able to communicate with the queue and
result-store, Redis or another network-accessible storage backend must be used.
.. note::
This section covers running multiple consumers against a *single* queue.
To run multiple *queues* (for example, to give different classes of
task their own worker pools), see :ref:`recipe-multiple-queues`.
.. _consumer-internals:
Consumer Internals
------------------
This section explains what happens when you call a ``task``-decorated function
in your application, by walking through the implementation of the consumer. The
`code for the consumer `_
is short (a couple hundred lines), and worth reading alongside this section.
The consumer is composed of three components: a master process, the scheduler,
and the worker(s). Depending on the worker type chosen, the scheduler and
workers will be run in their threads, processes or greenlets.
These three components coordinate the receipt, scheduling, and execution of
your tasks, respectively.
1. You call a function that huey has decorated, which triggers a message being
put into the queue (e.g a Redis list). At this point your application
returns immediately, returning a :py:class:`Result` object.
2. In the consumer process, the worker(s) will be listening for new messages
and one of the workers will receive your message indicating which task to
run, when to run it, and with what parameters.
3. The worker looks at the message and checks to see if it can be run (i.e.,
was this message "revoked"? Is it scheduled to actually run later?). If it
is revoked, the message is thrown out. If it is scheduled to run later, it
gets added to the schedule. Otherwise, it is executed.
4. The worker executes the task. If the task finishes, any results are stored
in the result store. If the task fails, the consumer checks to see if the
task can be retried. Then, if the task is to be retried, the consumer checks
to see if the task is configured to wait a number of seconds between
retries. Depending on the configuration, huey will either re-enqueue the
task for execution, or tell the scheduler when to re-enqueue it based on the
delay. If the consumer is killed abruptly or the machine powers off
unexpectedly, any tasks that are currently being run by a worker will be
"lost".
While all the above is going on with the Worker(s), the Scheduler is looking at
its schedule to see if any tasks are ready to be executed. If a task is ready
to run, it is enqueued and will be processed by the next available worker.
If you are using the Periodic Task feature (cron), then every minute, the
scheduler will check through the various periodic tasks to see if any should
be run. If so, these tasks are enqueued.
.. warning::
SIGINT is used to perform a graceful shutdown.
When the consumer is shutdown using SIGTERM, any workers still
involved in the execution of a task will be interrupted mid-task.
Signals
-------
The consumer will emit certain :ref:`signals` as it executes tasks. User code
can register signal handlers to respond to these events. For more information,
see the :ref:`signals` document.
---
## File: docs/contrib.rst
.. _contrib:
Huey Extensions
===============
The ``huey.contrib`` package contains modules that provide extra functionality
beyond the core APIs.
.. include:: django.rst
.. include:: asyncio.rst
.. include:: stats.rst
.. include:: flask_admin.rst
.. include:: mini.rst
---
## File: docs/django.rst
.. _django:
Django
------
.. image:: django-admin.png
Huey comes with special integration for use with the Django framework. The
integration provides:
1. Configuration of huey via the Django settings module.
2. Running the consumer as a Django management command.
3. Auto-discovery of ``tasks.py`` modules to simplify task importing.
4. Properly manage database connections.
5. A :ref:`backend ` for the ``django.tasks`` framework (Django
6.0 and newer, or older Djangos using the django-tasks backport).
6. Optional :ref:`Admin integration ` for visibility and
managing your Huey queue.
Supported Django versions are those officially supported at https://www.djangoproject.com/download/#supported-versions
Setting things up
^^^^^^^^^^^^^^^^^
To use huey with Django, the first step is to add an entry to your project's
``settings.INSTALLED_APPS``:
.. code-block:: python
# settings.py
# ...
INSTALLED_APPS = (
# ...
'huey.contrib.djhuey', # Add this to the list.
# ...
)
The above is the bare minimum needed to start using huey's Django integration.
If you like, though, you can also configure both Huey and the *consumer* using
the settings module.
Huey settings are optional. If not provided, Huey will default to using Redis
running on localhost:6379 (standard setup).
Configuration is kept in ``settings.HUEY``, which can be either a dictionary or
a :py:class:`Huey` instance. Here is an example that shows all of the supported
options with their default values:
.. code-block:: python
# settings.py
HUEY = {
'huey_class': 'huey.RedisHuey', # Huey implementation to use.
'name': settings.DATABASES['default']['NAME'], # Use db name for huey.
'results': True, # Store return values of tasks.
'store_none': False, # If a task returns None, do not save to results.
'immediate': settings.DEBUG, # If DEBUG=True, run synchronously.
'utc': True, # Use UTC for all times internally.
'blocking': True, # Perform blocking pop rather than poll Redis.
'connection': {
'host': 'localhost',
'port': 6379,
'db': 0,
'connection_pool': None, # Definitely you should use pooling!
# ... tons of other options, see redis-py for details.
# huey-specific connection parameters.
'read_timeout': 1, # If not polling (blocking pop), use timeout.
'url': None, # Allow Redis config via a DSN.
},
'consumer': {
'workers': 1,
'worker_type': 'thread',
'initial_delay': 0.1, # Smallest polling interval, same as -d.
'backoff': 1.15, # Exponential backoff using this rate, -b.
'max_delay': 10.0, # Max possible polling interval, -m.
'scheduler_interval': 1, # Check schedule every second, -s.
'periodic': True, # Enable crontab feature.
'check_worker_health': True, # Enable worker health checks.
'health_check_interval': 1, # Check worker health every second.
},
}
The following ``huey_class`` implementations are provided out-of-the-box:
* ``huey.RedisHuey`` - default.
* ``huey.PriorityRedisHuey`` - uses Redis but adds support for :ref:`priority`.
Requires redis server 5.0 or newer.
* ``huey.RedisExpireHuey`` - Redis implementation that expires result keys
automatically if results are not read.
* ``huey.PriorityRedisExpireHuey`` - Redis implementation that expires result
keys automatically if results are not read and supports priority.
* ``huey.SqliteHuey`` - uses Sqlite, full support for task priorities. Accepts
a ``filename`` parameter for the path to the database file.
* ``huey.CySqliteHuey`` - uses Sqlite via the `cysqlite `_
driver, full support for task priorities. Accepts a ``filename`` parameter
for the path to the database file, and an optional ``pragmas={...}`` dict of
connection settings.
* ``huey.PostgresHuey`` - uses Postgres, full support for task priorities. See
:ref:`django-postgres` for connection and schema configuration.
* ``huey.FileHuey`` - uses filesystem for storage. Accepts a ``path`` parameter
for the base storage directory.
Alternatively, you can set ``settings.HUEY`` to a :py:class:`Huey`
instance and do your configuration directly. In the example below, I've also
shown how you can create a connection pool:
.. code-block:: python
# settings.py: alternative configuration method
from huey import RedisHuey
from redis import ConnectionPool
pool = ConnectionPool(host='my.redis.host', port=6379, max_connections=20)
HUEY = RedisHuey('my-app', connection_pool=pool)
.. _django-postgres:
Using Postgres for storage
^^^^^^^^^^^^^^^^^^^^^^^^^^^
:py:class:`PostgresHuey` lets you use your existing Postgres database as the
task queue, schedule and result store (``pip install huey[postgres]``). Two
integration points are worth calling out for Django users.
Schema creation
"""""""""""""""
By default the storage backend issues ``create table if not exists`` for its
tables when it is instantiated. Because ``settings.HUEY`` is evaluated at import
time, that means every process importing ``huey.contrib.djhuey`` (including
your web workers) will connect and attempt to create the tables, which
requires the ``CREATE`` privilege and clashes with a migrations-managed schema.
Pass ``create_tables=False`` to disable the automatic DDL and create the tables
once, explicitly, using the ``create_huey_tables`` management command:
.. code-block:: python
# settings.py
HUEY = {
'huey_class': 'huey.PostgresHuey',
'create_tables': False,
'connection': {'dsn': 'postgresql:///my_db'},
}
.. code-block:: shell
./manage.py create_huey_tables
Reusing the Django database configuration
"""""""""""""""""""""""""""""""""""""""""
Rather than duplicating your credentials in ``settings.HUEY``, you can point the
storage at the same database Django uses by passing a ``connection`` callable.
The callable must return a **new**, dedicated ``psycopg`` connection. Do not
hand back ``django.db.connection``. Huey enables autocommit on the connection
and holds a long-lived one open for ``LISTEN``, neither of which is compatible
with Django's per-request connection or its transaction handling.
.. code-block:: python
# settings.py
import psycopg
from django.conf import settings
def huey_connection():
db = settings.DATABASES['default']
return psycopg.connect(
dbname=db['NAME'],
user=db.get('USER') or None,
password=db.get('PASSWORD') or None,
host=db.get('HOST') or None,
port=db.get('PORT') or None)
HUEY = {
'huey_class': 'huey.PostgresHuey',
'create_tables': False,
'connection': {'connection': huey_connection},
}
To ensure a task is not enqueued until the surrounding transaction commits (for
example, when it references a row created in the same view), use
:py:func:`on_commit_task` as described in :ref:`django-transactions`.
Running the Consumer
^^^^^^^^^^^^^^^^^^^^
To run the consumer, use the ``run_huey`` management command. This command
will automatically import any modules in your ``INSTALLED_APPS`` named
*tasks.py*. The consumer can be configured using both the django settings
module and/or by specifying options from the command-line.
Options specified on the command line take precedence over those specified in
the settings module.
To start the consumer, run:
.. code-block:: shell
./manage.py run_huey
In addition to the ``HUEY.consumer`` setting dictionary, the management command
supports all the same options as the standalone consumer. These options are
listed and described in the :ref:`Options for the consumer `
section.
For quick reference, the most important command-line options are briefly
listed here.
``-w``, ``--workers``
Number of worker threads/processes/greenlets. Default is 1, but most
applications should use at least 2.
``-k``, ``--worker-type``
Worker type, must be "thread", "process" or "greenlet". The default is
*thread*, which provides good all-around performance. For CPU-intensive
workloads, *process* is likely to be more performant. The *greenlet* worker
type is suited for IO-heavy workloads. When using *greenlet* you can
specify tens or hundreds of workers since they are extremely lightweight
compared to threads/processes. *See note below on using gevent/greenlet*.
``-A``, ``--disable-autoload``
Disable automatic loading of tasks modules.
Due to a conflict with Django's base option list, the "verbose" option is set
using ``-V`` or ``--huey-verbose``. When enabled, huey logs at the DEBUG level.
For more information, read the :ref:`Options for the consumer ` section.
Using gevent
^^^^^^^^^^^^
When using worker type *greenlet*, it's necessary to apply a monkey-patch
before any libraries or system modules are imported. Gevent monkey-patches
things like ``socket`` to provide non-blocking I/O, and if those modules are
loaded before the patch is applied, then the resulting code will execute
synchronously.
Unfortunately, because of Django's design, the only way to reliably apply this
patch is to create a custom bootstrap script that mimics the functionality of
``manage.py``. Here is the patched ``manage.py`` code:
.. code-block:: python
#!/usr/bin/env python
import os
import sys
# Apply monkey-patch if we are running the huey consumer.
if 'run_huey' in sys.argv:
from gevent import monkey
monkey.patch_all()
if __name__ == "__main__":
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "conf")
from django.core.management import execute_from_command_line
execute_from_command_line(sys.argv)
How to create tasks
^^^^^^^^^^^^^^^^^^^
The :py:meth:`~Huey.task` and :py:meth:`~Huey.periodic_task` decorators can be
imported from the ``huey.contrib.djhuey`` module. Here is how you might define
two tasks:
.. code-block:: python
from huey import crontab
from huey.contrib.djhuey import periodic_task, task
@task()
def count_beans(number):
print('-- counted %s beans --' % number)
return 'Counted %s beans' % number
@periodic_task(crontab(minute='*/5'))
def every_five_mins():
print('Every five minutes this will be printed by the consumer')
The ``huey.contrib.djhuey`` module exposes a number of additional helpers:
* :py:meth:`~Huey.lock_task`
* :py:meth:`~Huey.enqueue`
* :py:meth:`~Huey.restore`, :py:meth:`~Huey.restore_all`, :py:meth:`~Huey.restore_by_id`
* :py:meth:`~Huey.revoke`, :py:meth:`~Huey.revoke_all`, :py:meth:`~Huey.revoke_by_id`
* :py:meth:`~Huey.is_revoked`
* :py:meth:`~Huey.on_startup`
* :py:meth:`~Huey.pre_execute`
* :py:meth:`~Huey.post_execute`
* :py:meth:`~Huey.signal` and :py:meth:`~Huey.disconnect_signal`
* :py:func:`on_commit_task`, for enqueueing tasks after transaction commits.
Tasks that execute queries
^^^^^^^^^^^^^^^^^^^^^^^^^^
If you plan on executing queries inside your task, it is a good idea to close
the connection once your task finishes. To make this easier, huey provides a
special decorator to use in place of ``task`` and ``periodic_task`` which will
automatically close the connection for you.
.. code-block:: python
from huey import crontab
from huey.contrib.djhuey import db_periodic_task, db_task
@db_task()
def do_some_queries():
# This task executes queries. Once the task finishes, the connection
# will be closed.
@db_periodic_task(crontab(minute='*/5'))
def every_five_mins():
# This is a periodic task that executes queries.
.. py:function:: db_task(*args, **kwargs)
:param args: See :py:meth:`~Huey.task` for supported parameters.
:param kwargs: See :py:meth:`~Huey.task` for supported parameters.
Equivalent to :py:meth:`~Huey.task`, but closes the Django database
connections when the task finishes.
.. py:function:: db_periodic_task(validate_datetime, *args, **kwargs)
:param args: See :py:meth:`~Huey.periodic_task` for supported parameters.
:param kwargs: See :py:meth:`~Huey.periodic_task` for supported parameters.
Equivalent to :py:meth:`~Huey.periodic_task`, but closes the Django
database connections when the task finishes.
.. _django-transactions:
Enqueueing After Commit
^^^^^^^^^^^^^^^^^^^^^^^
The Huey Django extension provides a helper for when you need to ensure the
active transaction has committed **before** enqueueing your task. Consider this
example code:
.. code-block:: python
@task()
def do_work(user_id):
user = User.objects.get(pk=user_id)
...
@transaction.atomic
def some_view(request, ...):
user = User.objects.create(...)
do_work(user.id)
return response
When ``some_view`` is executed, the ``do_work()`` task is enqueued with the ID
of the newly-created user row. If the task is enqueued and executed **before**
the transaction commits, then the task will report that the user matching that
ID does not exist (because it has not been committed yet).
To avoid this situation, we provide a special-purpose decorator
:py:func:`on_commit_task`, which registers a callback with Django that ensures
the task is enqueued **after** the transaction is committed. If no transaction
is active, the task will be enqueued normally.
Here is the safe version:
.. code-block:: python
@on_commit_task()
def do_work(user_id):
user = User.objects.get(pk=user_id)
...
@transaction.atomic
def some_view(request, ...):
user = User.objects.create(...)
do_work(user.id) # Will not be enqueued until txn commits.
return response
Because we have to setup a callback to run after commit, the full functionality
of the :py:class:`TaskWrapper` is not available with tasks decorated with
:py:func:`on_commit_task`. If you anticipate needing all the TaskWrapper
methods, you can decorate the same function twice by given them two different
identifier names:
.. code-block:: python
def do_work(user_id):
user = User.objects.get(pk=user_id)
...
do_work_task = task(name="do_work_task")(do_work)
do_work_on_commit = on_commit_task(name="do_work_on_commit")(do_work)
.. py:function:: on_commit_task(*args, **kwargs)
:param args: See :py:meth:`~Huey.task` for supported parameters.
:param kwargs: See :py:meth:`~Huey.task` for supported parameters.
Enqueue the decorated function for execution after the transaction commits.
If no transaction is active, task will be enqueued immediately.
.. _django-admin-stats:
Admin dashboard
^^^^^^^^^^^^^^^
Huey can record task statistics (see :ref:`task-stats`) and surface them in
the Django admin. Enable it by adding the stats app to ``INSTALLED_APPS``:
.. code-block:: python
INSTALLED_APPS = [
# ...
'huey.contrib.djhuey',
'huey.contrib.djhuey.stats',
]
That is the whole setup. The app starts the stats recorder in every process
at start-up, including the consumer (``run_huey`` is a management command),
so both enqueue and execution events are captured with no further
configuration. The recorder uses `peewee `_ for
storage, creating its tables automatically in ``DATABASES['default']`` (there
are no migrations to run) and pruning old rows as new events are written.
The admin index gets a *Huey* section with two entries:
* **Dashboard**: live queue depths, a throughput chart, per-task statistics,
currently-running tasks and recent events, refreshed every 5 seconds, with
controls to revoke/restore tasks and flush the queue, schedule, results or
locks.
* **Events**: a filterable, searchable log of the raw signal events.
Options may be provided via a ``HUEY_STATS`` setting:
.. code-block:: python
HUEY_STATS = {
'retention_hours': 48, # Prune events older than this.
'max_events': 2000, # Cap on rows retained per queue.
'capture_args': False, # Store a truncated repr of task arguments.
# Stats are stored in DATABASES['default'] by default. Override with
# a peewee Database instance or db-url string:
#'database': 'sqlite:///huey-stats.db',
}
.. note::
If ``HUEY_STATS['database']`` points somewhere other than the default
Django database, the dashboard works as usual but the *Events* changelist
(which reads through the Django ORM) will not find the tables.
.. note::
The dashboard's *Registered tasks* table lists the tasks known to the
**web** process. ``run_huey`` autodiscovers each app's ``tasks`` module
but the web server does not, so import your tasks from your app's
``AppConfig.ready()`` to make them visible (see
``examples/django_ex``).
.. _django-task:
Django task framework
---------------------
Django 6.0 includes `django.tasks
`_, a standard
interface for defining and enqueueing background tasks. Django does not ship
a production backend for actually running them - huey provides one. Tasks
declared with the ``django.tasks.task`` decorator are stored and executed by
the regular huey consumer, side-by-side with your native huey tasks. On older
Django versions, the backend works identically with the `django-tasks
`_ backport package.
To use it, declare the backend in ``settings.TASKS``. The backend uses the
shared huey instance configured by ``settings.HUEY``, as described above:
.. code-block:: python
# settings.py
TASKS = {
'default': {
'BACKEND': 'huey.contrib.djhuey.tasks_backend.HueyBackend',
},
}
Tasks are declared and enqueued using the standard django.tasks APIs - no
huey imports are necessary:
.. code-block:: python
from django.tasks import task
@task
def send_welcome_email(user_id):
...
# In a view, etc:
result = send_welcome_email.enqueue(user.id)
result.id # Unique identifier, usable with get_result().
result.status # READY -> RUNNING -> SUCCESSFUL or FAILED.
# Later, check on it:
result.refresh()
if result.is_finished:
print(result.return_value)
The consumer is run exactly as before - ``manage.py run_huey`` - and executes
django.tasks tasks and native huey tasks side-by-side.
Supported functionality:
* ``run_after`` is mapped onto huey's ``eta`` and handled by the scheduler.
* ``priority`` is supported by every storage engine except ``RedisHuey`` and
``RedisExpireHuey``, where declaring a task with a non-zero priority raises
``InvalidTask``. Use ``PriorityRedisHuey`` or ``PriorityRedisExpireHuey``
with Redis.
* Results: ``get_result()``, ``refresh()``, return values and errors (with
tracebacks) are fully supported. Task status is tracked in the huey result
store, so ``RedisExpireHuey`` will expire result data automatically.
* Database connections are closed after each task, equivalent to
:py:func:`db_task`.
* Transactional enqueueing: add ``'ENQUEUE_ON_COMMIT': True`` to the backend
declaration to defer enqueueing until the active transaction commits
(equivalent to :py:func:`on_commit_task`).
* Immediate mode applies as usual: when ``DEBUG=True``, tasks execute
synchronously, through the same code-path the consumer uses.
.. note::
The django.tasks framework requires task functions to be defined at the
module level, and the huey backend additionally requires that they are
importable from their module path - lambdas cannot be enqueued.
Coroutine (``async def``) tasks are not supported.
DEBUG and Synchronous Execution
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
When ``settings.DEBUG = True``, and ``settings.HUEY`` is a ``dict`` that does
not explicitly specify a value for ``immediate``, tasks will be executed
**synchronously** just like regular function calls. The purpose of this is to
avoid running both Redis and an additional consumer process while developing or
running tests. If you prefer to use a live storage engine when ``DEBUG`` is
enabled, you can specify ``immediate_use_memory=False`` - which still runs Huey
in immediate mode, but using a live storage API. To completely disable
immediate mode when ``DEBUG`` is set, you can specify ``immediate=False`` in
your settings.
.. code-block:: python
# settings.py
HUEY = {
'name': 'my-app',
# To run Huey in "immediate" mode with a live storage API, specify
# immediate_use_memory=False.
'immediate_use_memory': False,
# OR:
# To run Huey in "live" mode regardless of whether DEBUG is enabled,
# specify immediate=False.
'immediate': False,
}
Getting the Huey Instance
^^^^^^^^^^^^^^^^^^^^^^^^^
If you want to interact with Huey APIs that are not exposed through ``djhuey``
explicitly, you can get the actual ``Huey`` instance in the following way:
.. code-block:: python
from huey.contrib.djhuey import HUEY as huey
# E.g., get the underlying Storage instance.
storage = huey.storage
Configuration Examples
^^^^^^^^^^^^^^^^^^^^^^
This section contains example ``HUEY`` configurations.
.. code-block:: python
# Redis running locally with four worker threads.
HUEY = {
'name': 'my-app',
'consumer': {'workers': 4, 'worker_type': 'thread'},
}
.. code-block:: python
# Redis on network host with 64 worker greenlets and connection pool
# supporting up to 100 connections.
from redis import ConnectionPool
pool = ConnectionPool(
host='192.168.1.123',
port=6379,
max_connections=100)
HUEY = {
'name': 'my-app',
'connection': {'connection_pool': pool},
'consumer': {'workers': 64, 'worker_type': 'greenlet'},
}
It is also possible to specify the connection using a Redis URL, making it easy
to configure this setting using a single environment variable:
.. code-block:: python
HUEY = {
'name': 'my-app',
'url': os.environ.get('REDIS_URL', 'redis://localhost:6379/?db=1')
}
Alternatively, you can just assign a :py:class:`Huey` instance to the ``HUEY`` setting:
.. code-block:: python
from huey import RedisHuey
HUEY = RedisHuey('my-app')
---
## File: docs/guide.rst
.. _guide:
Guide
=====
This document presents Huey using simple examples that cover the most common
usage of the library. Detailed documentation can be found in the
:ref:`API documentation `.
Example :py:meth:`~Huey.task` that adds two numbers:
.. code-block:: python
# demo.py
from huey import SqliteHuey
huey = SqliteHuey(filename='/tmp/demo.db')
@huey.task()
def add(a, b):
return a + b
To test, run the consumer, specifying the import path to the ``huey`` object:
.. code-block:: shell
huey_consumer demo.huey
In a Python shell, we can call our ``add`` task:
.. code-block:: pycon
>>> from demo import add
>>> r = add(1, 2)
>>> r()
3
If you try to resolve the result (``r``) before the task has been executed,
then ``r()`` will return ``None``. You can avoid this by instructing the
result to block until the task has finished and a result is ready:
.. code-block:: pycon
>>> r = add(1, 2)
>>> r(blocking=True, timeout=5) # Wait up to 5 seconds for result.
3
You can also check to see if a result is ready using :py:meth:`Result.is_ready`:
.. code-block:: pycon
>>> r = add(1, 2)
>>> r.is_ready()
False
>>> r.is_ready() # A few moments later.
True
>>> r()
3
What happens when we call a task function?
1. When the ``add()`` function is called, a message representing the call is
placed in a queue.
2. The function returns immediately without actually running, and returns a
:py:class:`Result` handle, which can be used to retrieve the result once the
task has been executed by the consumer.
3. The consumer process sees that a message has arrived, and a worker will call
the ``add()`` function and place the return value into the result store.
4. We can use the :py:class:`Result` handle to read the return value from the
result store.
For more information, see the :py:meth:`~Huey.task` decorator documentation.
Scheduling tasks
----------------
Tasks can be scheduled to execute at a certain time, or after a delay.
In the following example, we will schedule a call to ``add()`` to run in 10
seconds, and then will block until the result becomes available:
.. code-block:: pycon
>>> r = add.schedule((3, 4), delay=10)
>>> r(blocking=True) # Will block for ~10 seconds before returning.
7
If we wished to schedule the task to run at a particular time, we can use the
``eta`` parameter instead. The following example will run after a 10 second
delay:
.. code-block:: pycon
>>> eta = datetime.datetime.now() + datetime.timedelta(seconds=10)
>>> r = add.schedule((4, 5), eta=eta)
>>> r(blocking=True) # Will block for ~10 seconds.
9
What happens when we schedule a task?
1. When we call :py:meth:`~TaskWrapper.schedule`, a message is placed on the
queue instructing the consumer to call the ``add()`` function in 10 seconds.
2. The function returns immediately, and returns a :py:class:`Result` handle.
3. The consumer process sees that a message has arrived, and will notice that
the message is not yet ready to be executed, but should be run in ~10s.
4. The consumer adds the message to a schedule.
5. In ~10 seconds, the scheduler will pick-up the message and place it back
into the queue for execution.
6. A worker will dequeue the message, execute the ``add()`` function, and place
the return value in the result store.
7. The :py:class:`Result` handle from step 2 will now be able to read the
return value from the task.
For more details, see the :py:meth:`~TaskWrapper.schedule` API documentation.
Periodic tasks
--------------
Huey provides crontab-like functionality that enables functions to be executed
automatically on a given schedule.
In the following example, we will declare a :py:meth:`~Huey.periodic_task` that
executes every 3 minutes and prints a message on consumer process stdout:
.. code-block:: python
from huey import SqliteHuey
from huey import crontab
huey = SqliteHuey(filename='/tmp/demo.db')
@huey.task()
def add(a, b):
return a + b
@huey.periodic_task(crontab(minute='*/3'))
def every_three_minutes():
print('This task runs every three minutes')
Once a minute, the scheduler will check to see if any of the periodic tasks
should be called. If so, the task will be enqueued for execution.
Because periodic tasks are called independent of any user interaction, they
do not accept any arguments.
Similarly, the return-value for periodic tasks is discarded, rather than
being put into the result store. This is because there is not an obvious
way for an application to obtain a :py:class:`Result` handle to access the
result of a given periodic task execution.
The :py:func:`crontab` function accepts the following arguments, in the same
order as the standard Linux crontab format:
* minute
* hour
* day
* month
* day_of_week (0=Sunday, 6=Saturday)
Acceptable inputs:
* ``*`` - always true, e.g. if ``hour='*'``, then the rule matches any hour.
* ``*/n`` - every *n* interval, e.g. ``minute='*/15'`` means every 15 minutes.
* ``m-n`` - run every time ``m..n`` inclusive.
* ``m,n`` - run on *m* and *n*.
Multiple rules can be expressed by separating the individual rules with a
comma, for example:
.. code-block:: python
# Runs every 10 minutes between 9a and 11a, and 4p-6p.
crontab(minute='*/10', hour='9-11,16-18')
Huey also provides shortcuts for :py:func:`crontab.daily` and
:py:func:`crontab.hourly` schedules:
.. code-block:: python
@huey.periodic_task(crontab.hourly())
def aggregate_stats():
# This task runs every hour, on the hour.
...
For more information see the following API documentation:
* :py:meth:`~Huey.periodic_task`
* :py:func:`crontab`
Retrying tasks that fail
------------------------
Sometimes we may have a task that we anticipate might fail from time to time,
in which case we should retry it. Huey supports automatically retrying tasks a
given number of times, optionally with a delay between attempts.
Here we'll declare a task that fails approximately half of the time. To
configure this task to be automatically retried, use the ``retries`` parameter
of the :py:meth:`~Huey.task` decorator:
.. code-block:: python
import random
@huey.task(retries=2) # Retry the task up to 2 times.
def flaky_task():
if random.randint(0, 1) == 0:
raise Exception('failing!')
return 'OK'
What happens when we call this task?
1. Message is placed on the queue and a :py:class:`Result` handle is returned
to the caller.
2. Consumer picks up the message and attempts to run the task, but the call to
``random.randint()`` happens to return ``0``, raising an ``Exception``.
3. The consumer puts the error into the result store and the exception is
logged. If the caller resolves the :py:class:`Result` now, a
:py:class:`TaskException` will be raised which contains information about
the exception that occurred in our task.
4. The consumer notices that the task can be retried 2 times, so it decrements
the retry count and re-enqueues it for execution.
5. The consumer picks up the message again and runs the task. This time, the
task succeeds! The new return value is placed into the result store ("OK").
6. We can reset our :py:class:`Result` handle by calling
:py:meth:`~Result.reset` and then re-resolve it. The result handle will now
give us the new value, "OK".
Should the task fail on the first invocation, it will be retried up-to two
times. Note that it will be retried *immediately* after it returns.
To specify a delay between retry attempts, we can add a ``retry_delay``
argument. The task will be retried up-to two times, with a delay of 10 seconds
between attempts:
.. code-block:: python
@huey.task(retries=2, retry_delay=10)
def flaky_task():
# ...
To grow the delay after each failed attempt, add a ``retry_backoff``
multiplier. The task below is retried after 10 seconds, then 20, then 40:
.. code-block:: python
@huey.task(retries=3, retry_delay=10, retry_backoff=2)
def call_external_service():
# ...
.. note::
``retries``, ``retry_delay`` and ``retry_backoff`` can also be specified
for periodic tasks.
It is also possible to explicitly retry a task from within the task, by raising
a :py:class:`RetryTask` exception. When this exception is used, the task will
be retried regardless of whether it was declared with ``retries``. Similarly,
the task's remaining retries (if they were declared) will not be affected by
raising :py:class:`RetryTask`. Example:
.. code-block:: python
@huey.task()
def fetch_api_data(url):
try:
fh = urlopen(url)
except HTTPError:
# Try again in 60 seconds for an HTTP error (500, etc).
raise RetryTask(delay=60)
...
For more information, see the following API documentation:
* :py:meth:`~Huey.task` and :py:meth:`~Huey.periodic_task`
* :py:class:`Result`
* :py:class:`RetryTask`
Error handling
--------------
When a task raises an unhandled exception, Huey performs several actions:
1. The exception and traceback are logged.
2. An error result is stored in the result store (a dict containing the
exception representation, traceback, task ID, and remaining retries).
3. The ``SIGNAL_ERROR`` :ref:`signal ` is emitted.
4. If the task has retries remaining, it is re-enqueued (and
``SIGNAL_RETRYING`` is emitted).
5. If the task has an ``on_error`` handler, the error handler task is enqueued
with the exception as an argument.
Steps 2 and 5 run on every failed attempt by default. To defer them until a
task's retries are exhausted, see :ref:`store-intermediate-errors`.
Reading error results
^^^^^^^^^^^^^^^^^^^^^
When you call :py:meth:`Result.get` on a task that failed, a :py:class:`TaskException`
is raised. The ``TaskException`` has a ``metadata`` attribute containing details
about the failure:
.. code-block:: python
from huey.exceptions import TaskException
@huey.task()
def download(url):
resp = requests.get(url)
resp.raise_for_status()
return resp.text
result = download('https://example.com/missing')
try:
result.get(blocking=True, timeout=10)
except TaskException as exc:
print(exc.metadata['error']) # "HTTPError('404 ...')"
print(exc.metadata['traceback']) # Full traceback string.
print(exc.metadata['task_id']) # UUID of the failed task.
print(exc.metadata['retries']) # Retries remaining (0 if exhausted).
Result handle after a retry
^^^^^^^^^^^^^^^^^^^^^^^^^^^
When a task fails, the error result is written to the result store. If the task
is retried and eventually succeeds, the success result **overwrites** the error.
However, the :py:class:`Result` object caches the result locally after the
first read. To see the updated result after a retry, you must call
:py:meth:`~Result.reset` (only with the default
``store_intermediate_errors=True``, see :ref:`store-intermediate-errors`):
.. code-block:: python
@huey.task(retries=2, retry_delay=10)
def flaky_download(url):
resp = requests.get(url)
resp.raise_for_status()
return resp.text
result = flaky_download('https://example.com/intermittent')
try:
result.get(blocking=True, timeout=5)
except TaskException:
# First attempt failed. Wait for the retry to finish.
result.reset() # Clear cached error so we can read the new result.
value = result.get(blocking=True, timeout=30)
# value now contains the successful return value (or raises again).
.. _store-intermediate-errors:
Deferring errors until retries are exhausted
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
By default a failing task reports its error on every attempt, even those that
still have retries remaining. The exception is written to the result store, and
the ``on_error`` handler runs each time. This means a blocking
:py:meth:`Result.get` raises on the first failure. To read the result of a
later retry you have to call :py:meth:`~Result.reset` first, as shown above.
Pass ``store_intermediate_errors=False`` to change this. A failing task that
still has retries is re-enqueued, and nothing is written to the result store or
sent to the ``on_error`` handler. The error or return value is stored once the
task succeeds or runs out of retries.
.. code-block:: python
huey = RedisHuey('my-app', store_intermediate_errors=False)
@huey.task(retries=2, retry_delay=10)
def flaky_download(url):
resp = requests.get(url)
resp.raise_for_status()
return resp.text
result = flaky_download('https://example.com/x')
# Blocks until the task finishes. Returns the value on success, or raises
# a TaskException after the final attempt fails. No reset() needed.
value = result.get(blocking=True, timeout=60)
The ``SIGNAL_ERROR`` and ``SIGNAL_RETRYING`` :ref:`signals ` still fire
on every attempt, so you can still observe individual failures.
.. note::
This defaults to ``True`` for backwards compatibility. Setting it to
``False`` is recommended, and it may become the default in a future
release. It matches :py:class:`chord`, which already waits for retries to
run out before reporting an error.
For more information, see:
* :py:class:`Result`
* :py:class:`TaskException`
.. _immediate:
Immediate mode
--------------
Huey can be run in a special mode called *immediate* mode, which is very useful
during testing and development. In immediate mode, Huey will execute task
functions immediately rather than enqueueing them, while still preserving the
APIs and behaviors one would expect when running a dedicated consumer process.
Immediate mode can be enabled in two ways:
.. code-block:: python
huey = RedisHuey('my-app', immediate=True)
# Or at any time, via the "immediate" attribute:
huey = RedisHuey('my-app')
huey.immediate = True
To disable immediate mode:
.. code-block:: python
huey.immediate = False
By default, enabling immediate mode will switch your Huey instance to using
in-memory storage. This is to prevent accidentally reading or writing to live
storage while doing development or testing. If you prefer to use immediate mode
with live storage, you can specify ``immediate_use_memory=False`` when creating
your :py:class:`Huey` instance:
.. code-block:: python
huey = RedisHuey('my-app', immediate_use_memory=False)
You can try immediate mode in the Python shell. In the following example,
everything happens within the interpreter with no separate consumer process.
Because immediate mode switches to in-memory storage when enabled, no Redis
server is needed either:
.. code-block:: pycon
>>> from huey import RedisHuey
>>> huey = RedisHuey()
>>> huey.immediate = True
>>> @huey.task()
... def add(a, b):
... return a + b
...
>>> result = add(1, 2)
>>> result()
3
>>> add.revoke(revoke_once=True) # We can revoke tasks.
>>> result = add(2, 3)
>>> result() is None
True
>>> add(3, 4)() # No longer revoked, was restored automatically.
7
What happens if we try to schedule a task for execution in the future, while
using immediate mode?
.. code-block:: pycon
>>> result = add.schedule((4, 5), delay=60)
>>> result() is None # No result.
True
The task was not executed. Instead it was added to the in-memory storage
layer's schedule, which we can check by calling :py:meth:`Huey.scheduled`:
.. code-block:: pycon
>>> huey.scheduled()
[__main__.add: 8873...bcbd @2019-03-27 02:50:06]
Since immediate mode is fully synchronous, there is not a separate thread
monitoring the schedule. The schedule can still be read or written to, but
scheduled tasks will **not** automatically be executed and periodic tasks will
not be enqueued.
Testing Guidelines
------------------
When testing Huey task-decorated functions, a couple guidelines will make your
life easier.
1. Set your :py:class:`Huey` instance to :ref:`immediate`. Any code that calls
a task will run it synchronously, so it is safe to block on results as well.
2. Unit test individual tasks by using the ``.call_local()`` to call the
underlying function.
Examples:
.. code-block:: python
# Consider the following Huey instance and tasks:
huey = RedisHuey(...)
# Fake database to check side-effects.
database = []
@huey.task()
def add(a, b):
return a + b
@huey.periodic_task(crontab(...))
def run_reports():
global database
database.append(True) # Simulate a side-effect.
# The return values for periodic tasks are discarded when run by the
# Huey consumer. A return value may be helpful for testing, however, so
# in this example we will return something to demonstrate how to test
# our periodic task.
return 42
class TestMyTasks(unittest.TestCase):
def setUp(self):
# Make tasks run synchronously.
huey.immediate = True
def tearDown(self):
huey.immediate = False
def test_task(self):
result_handle = add(3, 4)
self.assertEqual(result_handle.get(), 7)
# Alternatively, you can also:
self.assertEqual(add.call_local(3, 4), 7)
def test_task_exceptions(self):
# We can also test exceptions.
result_handle = add(3, None) # Exception logged, but not raised.
with self.assertRaises(TaskException):
result_handle.get()
with self.assertRaises(TypeError):
# Exception raised directly when using call_local().
add.call_local(3, None)
def test_periodic_task(self):
# We cannot use the result-handle from a periodic task, because the
# results are always discarded by the consumer. In this case it is
# necessary to use `.call_local()` if we want to check the return
# value.
self.assertEqual(run_reports.call_local(), 42)
self.assertTrue(len(database), 1)
# If our periodic task has a side-effect, however, we can call it
# normally and check the side-effect happened. For example, if the
# run_reports() periodic task wrote a row to a database, we could
# do something like:
run_reports()
self.assertTrue(len(database), 2)
.. _priority:
Task priority
-------------
.. note::
Priority support for Redis requires Redis 5.0 or newer. To use task
priorities with Redis, use the :py:class:`PriorityRedisHuey` instead of
:py:class:`RedisHuey`.
Every other storage layer supports task priorities natively:
:py:class:`SqliteHuey`, :py:class:`CySqliteHuey`,
:py:class:`PostgresHuey`, the file-based :py:class:`FileHuey`, and the
in-memory storage layer used when :ref:`immediate` is enabled.
Huey tasks can be given a priority, allowing you to ensure that your most
important tasks do not get delayed when the workers are busy.
Priorities can be assigned to a task function, in which case all invocations of
the task will default to the given priority. Additionally, individual task
invocations can be assigned a priority on a one-off basis.
When no priority is given, the task will default to a priority of ``0``.
To see how this works, lets define a task that has a priority (``10``):
.. code-block:: python
@huey.task(priority=10)
def send_email(to, subj, body):
return mailer.send(to, 'webmaster@myapp.com', subj, body)
When we invoke this task, it will be processed *before* any other pending tasks
whose priority is less than 10. So we could imagine our queue looking something
like this:
* ``process_payment`` - priority = 50
* ``check_spam`` - priority = 1
* ``make_thumbnail`` - priority = 0 (default)
Invoke the ``send_email()`` task:
.. code-block:: python
send_email('new_user@foo.com', 'Welcome', 'blah blah')
Now the queue of pending tasks would be:
* ``process_payment`` - priority = 50
* ``send_email`` - priority = 10
* ``check_spam`` - priority = 1
* ``make_thumbnail`` - priority = 0
We can override the default priority by passing ``priority=`` as a keyword
argument to the task function:
.. code-block:: python
send_email('boss@mycompany.com', 'Important!', 'etc', priority=90)
Now the queue of pending tasks would be:
* ``send_email`` (to boss) - priority = 90
* ``process_payment`` - priority = 50
* ``send_email`` - priority = 10
* ``check_spam`` - priority = 1
* ``make_thumbnail`` - priority = 0
Task priority only affects the ordering of tasks as they are pulled from the
queue of pending tasks. If there are periods of time where your workers are not
able to keep up with the influx of tasks, Huey's ``priority`` feature can
ensure that your most important tasks do not get delayed.
Task-specific priority overrides can also be specified when scheduling a task
to run in the future:
.. code-block:: python
# Uses priority=10, since that was the default we used when
# declaring the send_email task:
send_email.schedule(('foo@bar.com', 'subj', 'msg'), delay=60)
# Override, specifying priority=50 for this task.
send_email.schedule(('bar@foo.com', 'subj', 'msg'), delay=60, priority=50)
Lastly, we can specify priority on :py:class:`~Huey.periodic_task`:
.. code-block:: python
@huey.periodic_task(crontab(minute='0', hour='*/3'), priority=10)
def some_periodic_task():
# ...
For more information:
* :py:class:`PriorityRedisHuey` - Huey implementation that adds support for
task priorities with the Redis storage layer.
* :py:class:`PostgresHuey`, :py:class:`SqliteHuey`, :py:class:`CySqliteHuey`,
:py:class:`FileHuey` and the in-memory storage used when immediate-mode is
enabled have full support for task priorities.
* :py:meth:`~Huey.task` and :py:meth:`~Huey.periodic_task`
Canceling or pausing tasks
--------------------------
Huey tasks can be cancelled dynamically at runtime. This applies to regular
tasks, tasks scheduled to execute in the future, and periodic tasks.
Any task can be canceled ("revoked"), provided the task has not started
executing yet. Similarly, a revoked task can be restored, provided it has not
already been processed and discarded by the consumer.
Using the :py:meth:`Result.revoke` and :py:meth:`Result.restore` methods:
.. code-block:: python
# Schedule a task to execute in 60 seconds.
res = add.schedule((1, 2), delay=60)
# Provided the 60s has not elapsed, the task can be canceled
# by calling the `revoke()` method on the result object.
res.revoke()
# We can check to see if the task is revoked.
res.is_revoked() # -> True
# Similarly, we can restore the task, provided the 60s has
# not elapsed (at which point it would have been read and
# discarded by the consumer).
res.restore()
To revoke *all* instances of a given task, use the
:py:meth:`~TaskWrapper.revoke` and :py:meth:`~TaskWrapper.restore` methods on
the task function itself:
.. code-block:: python
# Prevent all instances of the add() task from running.
add.revoke()
# We can check to see that all instances of the add() task
# are revoked:
add.is_revoked() # -> True
# We can enqueue an instance of the add task, and then check
# to verify that it is revoked:
res = add(1, 2)
res.is_revoked() # -> True
# To re-enable a task, we'll use the restore() method on
# the task function:
add.restore()
# Is the add() task enabled again?
add.is_revoked() # -> False
Huey provides APIs to revoke / restore on both individual instances of a task,
as well as all instances of the task. For more information, see the following
API docs:
* :py:meth:`Result.revoke` and :py:meth:`Result.restore` for revoking
individual instances of a task.
* :py:meth:`Result.is_revoked` for checking the status of a task instance.
* :py:meth:`TaskWrapper.revoke` and :py:meth:`TaskWrapper.restore` for revoking
all instances of a task.
* :py:meth:`TaskWrapper.is_revoked` for checking the status of the task
function itself.
Canceling from within a Task
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Huey provides a special :py:class:`CancelExecution` exception which can be
raised, either within a :py:meth:`~Huey.pre_execute` hook or within the body of
a :py:meth:`~Huey.task`-decorated function, to cancel the execution of the
task. Additionally, when raised from within a task, the ``CancelExecution``
exception can override the task's default retry policy, by specifying either
``retry=True/False``.
Example:
.. code-block:: python
@huey.task(retries=2)
def load_data():
if something_temporary_is_wrong():
# Task will be retried, even if it has run out of retries or is a
# task that does not specify any automatic retries.
raise CancelExecution(retry=True)
elif something_fatal_is_wrong():
# Task will NOT be retried, even if it has more than one retry
# remaining.
raise CancelExecution(retry=False)
elif cancel_and_maybe_retry():
# Task will only be retried if it has one or more retries
# remaining (this is the default).
raise CancelExecution()
...
For more information, see: :py:class:`CancelExecution`.
Canceling or pausing periodic tasks
-----------------------------------
The ``revoke()`` and ``restore()`` methods support some additional options
which may be especially useful for :py:meth:`~Huey.periodic_task`.
The :py:meth:`~TaskWrapper.revoke` method accepts two optional parameters:
* ``revoke_once`` - boolean flag, if set then only the next occurrence of the
task will be revoked, after which it will be restored automatically.
* ``revoke_until`` - datetime, which specifies the time at which the task
should be automatically restored.
For example, suppose we have a task that sends email notifications, but our
mail server goes down and won't be fixed for a while. We can revoke the task
for a couple of hours, after which time it will start executing again:
.. code-block:: python
@huey.periodic_task(crontab(minute='0', hour='*'))
def send_notification_emails():
# ... code to send emails ...
Here is how we might revoke the task for the next 3 hours:
.. code-block:: pycon
>>> now = datetime.datetime.now()
>>> eta = now + datetime.timedelta(hours=3)
>>> send_notification_emails.revoke(revoke_until=eta)
Alternatively, we could use ``revoke_once=True`` to just skip the next
execution of the task:
.. code-block:: pycon
>>> send_notification_emails.revoke(revoke_once=True)
At any time, the task can be restored using the usual
:py:meth:`~TaskWrapper.restore` method, and it's status can be checked using
the :py:meth:`~TaskWrapper.is_revoked` method.
Task expiration
---------------
Huey tasks can be configured with an expiration time. Setting an expiration
time on tasks will prevent them being run after the given time has elapsed.
Expiration times may be useful if your queue is busy and there may be a
significant lag between the time a task is enqueued and the time the consumer
starts executing it.
Expiration times can be specified as:
* ``datetime()`` instances, which are treated as absolute times.
* ``timedelta()`` or ``int``, which are relative to the time *at which the task
is enqueued*.
A default expire time can be provided when declaring a task:
.. code-block:: python
# Task must be executed by consumer within 60s of being enqueued.
@huey.task(expires=60)
def time_sensitive_task(...):
Expiration times can be specified per-invocation, as well:
.. code-block:: python
# Task must be executed by consumer within 5 minutes of being enqueued.
time_sensitive_task(report_file, expires=timedelta(seconds=300))
Expiration times can also be specified when scheduling tasks:
.. code-block:: python
# Task scheduled to run in 1 hour, and once enqueued for execution, must be
# run within 60 seconds.
time_sensitive_task.schedule(
args=(report_file,),
delay=timedelta(seconds=3600),
expires=timedelta(seconds=60))
# Example using absolute datetimes instead of relative deltas:
one_hr = datetime.now() + timedelta(seconds=3600)
time_sensitive_task.schedule(
args=(report_file,),
eta=one_hr,
expires=one_hr + timedelta(seconds=60))
.. _task-timeouts:
Task timeouts
-------------
Huey tasks can be interrupted by setting a ``timeout`` on the task, in seconds.
The actual implementation of the timeout mechanism depends on which worker type
you are using in the consumer, and is selected automatically by Huey:
* Process: uses ``SIGALRM``, most robust.
* Thread: threads do **NOT** support a hard timeout. To use timeouts with
threaded workers, do cooperative timeout-checking from within your task
function, described below.
* Greenlet: uses ``gevent.Timeout``, reliable so long as the task yields to the
event loop and is not held-up making a blocking call.
A default timeout can be provided when declaring a task:
.. code-block:: python
# Task must finish executing in 60s. If not, a `TaskTimeout` is raised and
# returned to the caller via the result handle.
@huey.task(timeout=60)
def maybe_slow(...):
Timeouts can be specified per-invocation, as well:
.. code-block:: python
# Task will raise a `TaskTimeout` if runs for more than 30s.
maybe_slow(report_file, timeout=30)
Timeouts can also be specified when scheduling tasks:
.. code-block:: python
maybe_slow.schedule(
args=(report_file,),
timeout=120)
When a task times-out, the ``SIGNAL_TIMEOUT`` :ref:`signal ` will be
fired.
.. _cooperative-timeout:
Cooperative Timeout
^^^^^^^^^^^^^^^^^^^
Threads do not support a hard timeout. To implement timeouts with threaded
workers, or if you prefer more control within your task function, use the
cooperative timeout APIs:
* :py:meth:`Task.check_timeout`
* :py:attr:`Task.is_timed_out`
* :py:attr:`Task.time_remaining`
These methods require that the :py:class:`Task` instance be passed as context
into the task function with ``context=True``.
Example:
.. code-block:: python
@huey.task(timeout=60, context=True) # 60s deadline, takes Task as context.
def process_report(data, task=None):
for batch in chunk(data, 100):
task.check_timeout() # Will trigger a TaskTimeout if exceeds 60s.
service.process_data(batch) # Some expensive/slow operation.
For budget-aware scheduling you can use the more granular ``time_remaining``:
.. code-block:: python
@huey.task(timeout=60, context=True) # 60s deadline, takes Task as context.
def process_accounts(accounts, task=None):
for i, account in enumerate(accounts):
# If it looks like we might run out of time, schedule the remaining
# accounts to run in 5 minutes.
if task.time_remaining < 10:
process_accounts.schedule(
args=(accounts[i:],),
delay=300)
return
service.process_single_account(account)
Locking tasks
-------------
Task locking can be accomplished using the :py:meth:`Huey.lock_task` method,
which can be used as a context-manager or decorator.
This lock prevents multiple invocations of a task from running concurrently.
If a second invocation occurs and the lock cannot be acquired, then a special
:py:class:`TaskLockedException` is raised and the task will not be executed.
If the task is configured to be retried, then it will be retried normally.
Examples:
.. code-block:: python
@huey.periodic_task(crontab(minute='*/5'))
@huey.lock_task('reports-lock') # Goes *after* the task decorator.
def generate_report():
# If a report takes longer than 5 minutes to generate, we do
# not want to kick off another until the previous invocation
# has finished.
run_report()
@huey.periodic_task(crontab(minute='0'))
def backup():
# Generate backup of code
do_code_backup()
# Generate database backup. Since this may take longer than an
# hour, we want to ensure that it is not run concurrently.
with huey.lock_task('db-backup'):
do_db_backup()
See :py:meth:`Huey.lock_task` for API documentation.
.. _rate-limiting:
Rate-Limiting
-------------
Huey provides simple fixed-window rate-limiting for tasks. Rate-limiting
behaves much like locking and can be used as a context-manager or a decorator.
Rate-limiting allows for granular specification of when and how a rate-limited
task should be retried.
Example:
.. code-block:: python
@huey.task()
@huey.rate_limit('data-sync', limit=10, per=60)
def data_sync(data):
# synchronize `data` with another system, but limit to 10 invocations
# in a 60s window.
service.apply_updates(data)
In the above example, if more than 10 calls to ``data_sync()`` occur during a
60 second window, the task raises a :py:class:`RateLimitExceeded`. This
exception, by default, will cause the task to be retried at the start of the
next rate-limit window - in the example this would be some delay between 0 and
60 seconds. This occurs even if the task has **no** retries, as above.
Flow:
* ``data_sync()`` called 10 times in rapid succession, all succeed.
* ``data_sync()`` called again 33 seconds into the rate-limit window.
* ``RateLimitExceeded`` is raised. Huey schedules the rate-limited 11th
invocation of ``data_sync()`` to run in 27 seconds when the window resets.
* ``data_sync()`` called again 48 seconds into the rate-limit window.
* ``RateLimitExceeded`` raised, 12th invocation of ``data_sync`` scheduled to
run in 12 seconds when the window resets.
This is good for tasks that you must not lose, but which may overwhelm a
service if they come in too quickly. There is a risk, however, if the tasks
continually exceed the rate-limits, that they "outpace" the rate-limiting and
pile up. This can be mitigated by specifying a limited number of retries on the
task itself:
Example:
.. code-block:: python
@huey.task(retries=2)
@huey.rate_limit('no-thundering-herd', limit=10, per=60, retry=False)
def data_sync(data):
service.apply_updates(data)
This example will behave like the previous example, but will cap each
invocation to two retries.
Equivalent example using a context-manager instead of a decorator:
.. code-block:: python
@huey.task(retries=2)
def data_sync(data):
with huey.rate_limit('data-sync', limit=10, per=60, retry=False):
service.apply_updates(data)
See :py:meth:`Huey.rate_limit` for API documentation.
Task pipelines
--------------
Huey supports pipelines (or chains) of one or more tasks that should be
executed sequentially.
To get started, let's review the usual way we execute tasks:
.. code-block:: python
@huey.task()
def add(a, b):
return a + b
result = add(1, 2)
An equivalent, but more verbose, way is to use the :py:meth:`~TaskWrapper.s`
method to create a :py:class:`Task` instance and then enqueue it explicitly:
.. code-block:: python
# Create a task representing the execution of add(1, 2).
task = add.s(1, 2)
# Enqueue the task instance, which returns a Result handle.
result = huey.enqueue(task)
So the following are equivalent:
.. code-block:: python
result = add(1, 2)
# And:
result = huey.enqueue(add.s(1, 2))
The :py:meth:`TaskWrapper.s` method is used to create a :py:class:`Task`
instance (which represents the execution of the given function). The
``Task`` is what gets serialized and sent to the consumer.
To create a pipeline, we will use the :py:meth:`TaskWrapper.s` method to create
a :py:class:`Task` instance. We can then chain additional tasks using the
:py:meth:`Task.then` method:
.. code-block:: python
add_task = add.s(1, 2) # Create Task to represent add(1, 2) invocation.
# Add additional tasks to pipeline by calling add_task.then().
pipeline = (add_task
.then(add, 3) # Call add() with previous result (1+2) and 3.
.then(add, 4) # Previous result ((1+2)+3) and 4.
.then(add, 5)) # Etc.
# When a pipeline is enqueued, a ResultGroup is returned (which is
# comprised of individual Result instances).
result_group = huey.enqueue(pipeline)
# Print results of above pipeline.
print(result_group.get(blocking=True))
# [3, 6, 10, 15]
# Alternatively, we could have iterated over the result group:
for result in result_group:
print(result.get(blocking=True))
# 3
# 6
# 10
# 15
When enqueueing a task pipeline, the return value will be a
:py:class:`ResultGroup`, which encapsulates the :py:class:`Result` objects for
the individual tasks. :py:class:`ResultGroup` can be iterated over or you can
use the :py:meth:`ResultGroup.get` method to get all the task return values as
a list.
Note that the return value from the parent task is passed to the next task in
the pipeline, and so on.
If the value returned by the parent function is a ``tuple``, then the tuple
will be used to extend the ``*args`` for the next task. Likewise, if the
parent function returns a ``dict``, then the dict will be used to update the
``**kwargs`` for the next task.
Example of chaining fibonacci calculations:
.. code-block:: python
@huey.task()
def fib(a, b=1):
a, b = a + b, a
return (a, b) # returns tuple, which is passed as *args
pipe = (fib.s(1)
.then(fib)
.then(fib)
.then(fib))
results = huey.enqueue(pipe)
print(results(True)) # Resolve results, blocking until all are finished.
# [(2, 1), (3, 2), (5, 3), (8, 5)]
For more information, see the following API docs:
* :py:meth:`TaskWrapper.s`
* :py:meth:`Task.then`
* :py:class:`ResultGroup` and :py:class:`Result`
.. warning::
If a pipeline step returns ``None`` and :py:class:`Huey` was initialized
with ``store_none=False`` (the default), the result for that step will
not be written to the result store. When you read the :py:class:`ResultGroup`,
that step's value will appear as ``None`` and be indistinguishable from
"result not ready yet."
Pipeline *execution* is unaffected: the next task in the chain still runs,
but it receives no additional arguments from the ``None``-returning step
(since ``None`` is treated as "no data to pass").
If your pipeline steps may legitimately return ``None`` and you need to
read results from the :py:class:`ResultGroup`, initialize Huey with
``store_none=True`` or (preferable) have your tasks return a sentinel value
instead.
Error pipelines
^^^^^^^^^^^^^^^
Just as :py:meth:`Task.then` chains a follow-up task on success, the
:py:meth:`Task.error` method chains a task that runs when the parent fails.
The exception is passed as the first argument to the error handler:
.. code-block:: python
@huey.task()
def download(url):
resp = requests.get(url)
resp.raise_for_status()
return resp.text
@huey.task()
def on_download_error(exc, url=None):
logger.error('Download of %s failed: %s', url, exc)
send_alert('Download failed', str(exc))
# Build a task with an error handler.
task = download.s('https://example.com/data')
task.error(on_download_error, url='https://example.com/data')
result = huey.enqueue(task)
Error handlers can also be combined with success pipelines:
.. code-block:: python
task = (download.s('https://example.com/data')
.then(parse_response)
.then(store_result))
task.error(on_download_error)
result = huey.enqueue(task)
If ``download`` raises an exception, ``on_download_error`` is called and the
success pipeline (``parse_response``, ``store_result``) is skipped. If
``download`` succeeds, ``parse_response`` runs and ``on_download_error`` is
never called.
.. note::
The error handler is attached to the first task in the pipeline. If a
*later* task in the pipeline fails (e.g., ``parse_response``), the error
handler on ``download`` is not triggered. To handle errors in later stages,
attach error handlers to those tasks directly.
For more information, see:
* :py:meth:`Task.error`
* :py:class:`TaskException`
.. _groups-and-chords:
Groups and Chords
-----------------
Pipelines execute tasks sequentially, while groups and chords allow you to run
tasks in parallel, optionally collecting their results and passing them to a
final callback.
group - parallel execution
^^^^^^^^^^^^^^^^^^^^^^^^^^
:py:class:`group` allows you to enqueue one or more tasks in parallel and
gather the results. All tasks are enqueued immediately and may execute
concurrently across workers. A :py:class:`ResultGroup` is returned to enable
fetching the results of the tasks individually, as they become available, or
after they are all completed.
.. code-block:: python
from huey import group
@huey.task()
def fetch(url):
return requests.get(url).json()
urls = ['https://.../1/', 'https://.../2/', ...]
result_group = huey.enqueue(group([
fetch.s(url) for url in urls]))
# Block until all tasks are done.
results = result_group.get(blocking=True)
# [{'id': 1, ...}, {'id': 2, ...}, ...]
Unlike :py:meth:`~TaskWrapper.map`, groups may contain different types of
tasks:
.. code-block:: python
@huey.task()
def check_cache():
return cache.ping()
@huey.task()
def check_db():
try:
return db.execute('select 1').fetchone() is not None
except Exception:
logger.exception('database health-check failed')
return False
g = group([check_cache.s(), check_db.s()])
result_group = huey.enqueue(g)
cache_ok, db_ok = result_group(blocking=True)
Error handlers can be attached to every member of a group:
.. code-block:: python
@huey.task()
def on_fetch_error(exc):
logger.error('Fetch failed: %s', exc)
g = group([fetch.s(u) for u in urls]).error(on_fetch_error)
chord - map / reduce
^^^^^^^^^^^^^^^^^^^^
:py:class:`chord` is a :py:class:`group` with a callback. The member tasks are
run in parallel and when all tasks are finished, their results are passed to
the callback task in order.
.. code-block:: python
from huey import chord
@huey.task()
def fetch(url):
return (url, requests.get(url).text)
@huey.task()
def index_pages(results):
for url, html in results:
search.index(url, html)
return len(results)
# Huey will run all the fetch_url() tasks, then when they are all finished
# the `index_pages()` task will be enqueued with the task results.
urls = ['https://a.com', 'https://b.com', ...]
c = chord(
[fetch.s(url) for url in urls],
index_pages)
result = huey.enqueue(c)
pages = result(blocking=True)
print('Indexed %s pages' % pages)
What happens when a :py:class:`chord` is enqueued?
1. All sub-tasks are placed on the queue. Workers begin executing them as they
are dequeued.
2. As each sub-task completes (or fails permanently), its return value (or
exception) is stored and an internal completion counter is atomically
incremented.
3. When the last sub-task is complete, all sub-task results are collected in
order. The final callback is then enqueued with the sub-task results.
4. The callback is executed by a worker and the final result is made available.
.. note::
A sub-task that is skipped without executing (because it was revoked,
expired, or cancelled by a pre-execute hook) still counts towards chord
completion and contributes ``None`` as its result. A sub-task that is
interrupted by an abrupt consumer shutdown, however, is lost (tasks are
delivered at-most-once), in which case the callback will not fire.
Enqueueing a :py:class:`chord` returns a :py:class:`ChordResult`, which
provides access to the final callback result, the sub-task results, and any
tasks chained to the final callback.
.. code-block:: python
c = chord([fetch.s(u) for u in urls], index_pages).then(send_report)
result = huey.enqueue(c)
result() # Get result of index_pages (chord callback).
result.results # ResultGroup for the sub-tasks (fetch(...)).
result.callback # Result handle for callback.
result.pipeline_results # ResultGroup for [index_pages, send_report]
Composing groups and chords
^^^^^^^^^^^^^^^^^^^^^^^^^^^
Groups and chords allow the same composition APIs as pipelines. The following
are equivalent:
.. code-block:: python
@huey.task()
def incr(n):
return n + 1
@huey.task()
def total(ns):
return sum(ns)
result = huey.enqueue(
group([incr.s(i) for i in range(10)])
.then(total))
result = huey.enqueue(
chord([incr.s(i) for i in range(10)],
total))
Using :py:meth:`~chord.then` or :py:meth:`~chord.error` on a :py:class:`chord`
will attach an additional callback to the chord's callback:
.. code-block:: python
result = huey.enqueue(
group([incr.s(i) for i in range(3)])
.then(total)
.then(incr)
.error(alert_admin))
# Chord result is the output of total().
print(result(blocking=True)) # 1 + 2 + 3 == 6.
# Results of chord pipeline are 6, 7.
print(result.pipeline_results(blocking=True)) # [6, 7]
In this example, ``total`` receives the results from the ``incr`` subtasks.
Then the result of ``total`` is passed to ``incr`` again. If ``total``
raises an exception, ``alert_admin`` is executed instead.
Execution flow:
1. ``incr(0)``, ``incr(1)`` and ``incr(2)`` are enqueued.
2. All three eventually finish and ``total([1, 2, 3])`` is enqueued.
3. ``incr(6)`` is then enqueued and executed, returning ``7``.
Pipelines inside chords
^^^^^^^^^^^^^^^^^^^^^^^
Individual chord members can be pipelines:
.. code-block:: python
@huey.task()
def incr(n):
return n + 1
@huey.task()
def total(ns):
return sum(ns)
result = huey.enqueue(chord(
[incr.s(i).then(incr).then(incr) for i in range(3)],
total))
# First runs incr(0), incr(1), incr(2).
# Then: runs incr(1), incr(2), incr(3).
# Then: runs incr(2), incr(3), incr(4).
# Then: runs total([3, 4, 5])
print(result(True)) # 12.
The chord callback is not enqueued until all sub-task pipelines have completed.
chords inside chords
^^^^^^^^^^^^^^^^^^^^
chords can also contain chords:
.. code-block:: python
@huey.task()
def fetch(url):
return requests.get(url).json()
@huey.task()
def merge(results):
combined = {}
for r in results:
combined.update(r)
return combined
@huey.task()
def publish(merged):
database.write(merged)
inner1 = chord([fetch.s(u) for u in us_urls], merge.s())
inner2 = chord([fetch.s(u) for u in eu_urls], merge.s())
result = huey.enqueue(
chord([inner1, inner2], publish.s()))
In this example, two inner chords run in parallel. Each inner chord fans out
to its own set of URLs, then merges the results. When **both** inner merge
tasks complete, their results are collected and passed to ``publish``.
Error handling in chords
^^^^^^^^^^^^^^^^^^^^^^^^
chords are designed to **always complete**. When a sub-task fails and has no
retries remaining, the exception is sent to the results list in place of a
normal return value. The callback fires once all sub-tasks have completed, and
in the event an error occurred, that sub-tasks' value is the exception.
This means the callback may receive a mix of normal values and exceptions:
.. code-block:: python
@huey.task()
def aggregate(results):
if any(isinstance(r, Exception) for r in results):
logger.warning('One of the sub-tasks failed.')
return process([r for r in results if not isinstance(r, Exception)])
If a sub-task fails but has retries remaining, it will retry normally. The
chord will not receive an Exception unless the sub-task has exhausted available
retries.
.. code-block:: python
@huey.task(retries=3, retry_delay=10)
def flaky_fetch(url):
return requests.get(url).json()
c = chord(
[flaky_fetch.s(u) for u in urls],
aggregate.s())
If you prefer to handle errors before they reach the callback, catch
exceptions inside your sub-tasks and return a sentinel value:
.. code-block:: python
@huey.task()
def safe_fetch(url):
try:
return requests.get(url).json()
except Exception:
return None
@huey.task()
def aggregate(results):
valid = [r for r in results if r is not None]
return process(valid)
Error callbacks on members work independently of the chord and can be
applied via :py:meth:`group.error`:
.. code-block:: python
@huey.task()
def on_fetch_error(exc):
logger.error('Fetch failed: %s', exc)
tasks = [fetch.s(u).error(on_fetch_error) for u in urls]
c = chord(tasks, aggregate.s())
# Or equivalently using group.error():
c = (group([fetch.s(u) for u in urls])
.error(on_fetch_error)
.then(aggregate))
Error callbacks on the chord itself apply to the *callback* task:
.. code-block:: python
c = chord(
[fetch.s(u) for u in urls],
aggregate.s()
).error(alert_admin)
In this example, ``alert_admin`` runs if ``aggregate`` raises an exception,
*not* if a member fails.
.. note::
Revoking a chord member will prevent it from running, but the chord will
still complete: the revoked member contributes ``None`` to the results
and the callback fires normally. If you need to cancel a chord, revoke
the callback task instead.
.. note::
Unlike regular tasks, chord sub-task results are stored unconditionally
by the chord's internal bookkeeping. If a sub-task returns ``None``, the
callback will correctly receive ``None`` in the results list regardless
of the ``store_none`` setting. You do not need ``store_none=True`` for
chords to work with ``None`` return values.
Dynamic fan-out
^^^^^^^^^^^^^^^
Because chord members must be known at enqueue time, dynamic fan-out (where
the set of subtasks depends on a runtime value) should be done from inside
a task:
.. code-block:: python
@huey.task()
def discover_and_fetch(config):
urls = compute_urls(config)
result = huey.enqueue(chord(
[fetch.s(u) for u in urls],
aggregate.s()))
return result.callback.id # Return the callback task id for tracking.
For more information, see the following API docs:
* :py:class:`group`
* :py:class:`chord`
* :py:class:`ResultGroup` and :py:class:`ChordResult`
Signals
-------
The :py:class:`Consumer` sends :ref:`signals ` as it processes tasks.
The :py:meth:`Huey.signal` method can be used to attach a callback to one or
more signals, which will be invoked synchronously by the consumer when the
signal is sent.
As an example, we can add a signal handler that prints the signal name and the
ID of the related task.
.. code-block:: python
@huey.signal()
def print_signal_args(signal, task, exc=None):
if signal == SIGNAL_ERROR:
print('%s - %s - exception: %s' % (signal, task.id, exc))
else:
print('%s - %s' % (signal, task.id))
The :py:meth:`~Huey.signal` method is used to decorate the signal-handling
function. It accepts an optional list of signals. If none are provided, as in
our example, then the handler will be called for any signal.
The callback function (``print_signal_args``) accepts two required arguments,
which are present on every signal: ``signal`` and ``task``. Additionally, our
handler accepts an optional third argument ``exc`` which is only included with
``SIGNAL_ERROR``. ``SIGNAL_ERROR`` is only sent when a task raises an uncaught
exception during execution.
.. warning::
Signal handlers are executed *synchronously* by the consumer, so it is
typically a bad idea to introduce any slow operations into a signal
handler.
Huey emits the following signals:
* ``SIGNAL_ENQUEUED`` - task has been placed on the queue.
* ``SIGNAL_EXECUTING`` - task is about to be executed by a worker.
* ``SIGNAL_COMPLETE`` - task finished successfully.
* ``SIGNAL_ERROR`` - task raised an unhandled exception. Handler receives an
extra ``exc`` argument containing the exception instance.
* ``SIGNAL_CANCELED`` - task was canceled via :py:class:`CancelExecution`.
* ``SIGNAL_RETRYING`` - task failed but will be retried.
* ``SIGNAL_SCHEDULED`` - task was added to the schedule for future execution.
* ``SIGNAL_REVOKED`` - task was revoked and will not be executed.
* ``SIGNAL_EXPIRED`` - task's expiration time passed before execution.
* ``SIGNAL_LOCKED`` - task could not acquire its lock.
* ``SIGNAL_TIMEOUT`` - task exceeded its execution timeout.
* ``SIGNAL_RATE_LIMITED`` - task was rate-limited.
* ``SIGNAL_INTERRUPTED`` - consumer was shut down while the task was still
executing.
All signals are available in the ``huey.signals`` module. For signal ordering,
registration details, and examples, see the :ref:`signals` document and the
:py:meth:`Huey.signal` API documentation.
Consumer shutdown and interrupted tasks
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
When the consumer is shut down gracefully (``SIGINT`` / Ctrl+C), it finishes
executing any in-progress tasks before exiting. However, if the consumer is
killed with ``SIGTERM`` or crashes, tasks that are mid-execution are lost and
will not be retried automatically.
Huey tracks in-flight tasks internally. When the consumer shuts down, it calls
:py:meth:`Huey.notify_interrupted_tasks`, which emits ``SIGNAL_INTERRUPTED``
for each task that was still executing. You can use this signal to re-enqueue
interrupted tasks:
.. code-block:: python
from huey.signals import SIGNAL_INTERRUPTED
@huey.signal(SIGNAL_INTERRUPTED)
def on_interrupted(signal, task, *args, **kwargs):
# Re-enqueue the task so it will be executed again.
huey.enqueue(task)
.. warning::
Re-enqueueing interrupted tasks means the task may run more than once.
If your task is not idempotent, consider tracking execution state
externally (e.g. in a database) rather than blindly re-enqueueing.
.. _logging:
Logging
-------
Huey uses the standard library ``logging`` module to log information about task
execution and consumer activity. Messages are logged to the ``huey`` namespace,
with consumer-specific messages being logged to ``huey.consumer``.
When the consumer is run, it binds a default ``StreamHandler()`` to the huey
namespace so that all messages are logged to the console. The consumer logging
can be configured using the following consumer options:
* ``-l FILE, --logfile=FILE`` - log to a file.
* ``-v, --verbose`` - verbose logging (includes DEBUG level)
* ``-q, --quiet`` - minimal logging
* ``-S, --simple`` - simple logging format ("time message")
If you would like to get email alerts when an error occurs, you can attach a
``logging.handlers.SMTPHandler`` to the ``huey`` namespace at level ``ERROR``:
.. code-block:: python
from logging.handlers import SMTPHandler
import logging
mail_handler = SMTPHandler(
mailhost=('smtp.gmail.com', 587),
fromaddr='errors@myapp.com',
toaddrs=['developers@myapp.com'],
subject='Huey error log',
credentials=('errors@myapp.com', 'secret_password'),
secure=())
mail_handler.setLevel(logging.ERROR)
logging.getLogger('huey').addHandler(mail_handler)
.. _storage_options:
Storage Options
---------------
Huey provides a number of different storage layers suitable to different types
of workloads. Below I will try to sketch the differences, strengths, and
weaknesses of each storage layer.
:py:class:`RedisHuey`
Huey's capabilities are, to a large extent, informed by the functionality
available in Redis. This is the most robust option available and can handle
very busy workloads. Because Redis runs as a separate server process, it is
even possible to run Huey consumers on multiple machines to facilitate
"scale-out" operation.
Operations are guaranteed to be atomic, following the guarantees provided
by Redis. The queue is stored in a Redis list, scheduled tasks use a sorted
set, and the task result-store is kept in a hash.
Tasks that return a meaningful value rely on the caller "resolving" those
return values at some point, so the result store does not fill up with
unused data. Tasks whose result you never read should return ``None``.
By default Huey performs a "blocking" pop on the queue, which reduces
latency, although polling can be used instead by passing ``blocking=False``
when instantiating ``RedisHuey``.
For low-latency result-fetching, you can specify ``notify_result=True``
when instantiating ``RedisHuey``. This works by coordinating result
readiness through a blocking pop on a dedicated result list. By default
this result-readiness list will expire after 86400 seconds, but this can be
controlled with the ``notify_result_ttl`` parameter. **All RedisHuey**
implementations support these options.
Task priorities are not supported by :py:class:`RedisHuey`.
:py:class:`PriorityRedisHuey`
Redis storage layer that supports task priorities. To make this efficient,
``PriorityRedisHuey`` stores the queue in a sorted set. Since sorted sets
require unique keys, Huey uses the timestamp in microseconds to
differentiate tasks enqueued with the same priority.
:py:class:`RedisExpireHuey`
Redis storage layer that stores each task result in its own top-level key
with an expiration time, so the result-store does not fill up with
unresolved result values. The default expire time is 86400 seconds, set
with the ``expire_time`` parameter during instantiation.
:py:class:`PriorityRedisExpireHuey`
Combines the behaviors of :py:class:`PriorityRedisHuey` to support task
priorities, with the result-store expiration behavior of
:py:class:`RedisExpireHuey`.
:py:class:`SqliteHuey`
Sqlite works well for many workloads (see `Appropriate uses for Sqlite `_),
and Huey's Sqlite storage layer works well regardless of the worker-type
chosen. Sqlite locks the database during writes, ensuring only a single
writer can write to the database at any given time. Writes generally happen
very quickly, however, so in practice this is rarely an issue. Because the
database is stored in a single file, taking backups is simple.
``SqliteHuey`` may be a good choice for moderate workloads where the
operational complexity of running a separate server process like Redis is
undesirable.
:py:class:`CySqliteHuey`
Same storage layout and characteristics as :py:class:`SqliteHuey`, but
driven by `cysqlite `_ rather than the
standard library ``sqlite3`` module. Connection tuning is done with an
open-ended ``pragmas`` dict instead of a fixed set of parameters.
:py:class:`PostgresHuey`
Workers dequeue using ``select ... for update skip locked``, so any number
of consumers, on any number of hosts, can share a queue without contending
for the same rows. ``LISTEN``/``NOTIFY`` wakes an idle worker as soon as a
task is enqueued, so there is no polling latency. Huey's four tables can
live in the application's existing database, which means the queue is
backed up and restored along with everything else.
Each worker holds one additional connection for ``LISTEN``, so a consumer
with N workers uses N+1 connections. Connection poolers that multiplex
(PgBouncer in transaction-pooling mode) break ``LISTEN``. Initialize with
``blocking=False`` to fall back to polling in that case.
``PostgresHuey`` is a good choice when your application already runs
Postgres and you would rather not operate a separate Redis server.
:py:class:`FileHuey`
Stores the queue, schedule and task results in files on the filesystem.
This implementation is provided mostly for testing and development. An
exclusive lock is used around all file-system operations, since multiple
operations (list directory, read file, unlink file, e.g.) are typically
required for each storage primitive (enqueue, dequeue, store result, etc).
:py:class:`MemoryHuey`
In-memory implementation of the storage layer used for :ref:`immediate`.
:py:class:`BlackHoleHuey`
All storage methods are no-ops.
Tips and tricks
---------------
To call a task-decorated function in its original form, you can use
:py:meth:`~TaskWrapper.call_local`:
.. code-block:: python
@huey.task()
def add(a, b):
return a + b
# Call the add() function in "un-decorated" form, skipping all
# the huey stuff:
add.call_local(3, 4) # Returns 7.
It's also worth mentioning that python decorators are just syntactical sugar
for wrapping a function with another function. Thus, the following two examples
are equivalent:
.. code-block:: python
@huey.task()
def add(a, b):
return a + b
# Equivalent to:
def _add(a, b):
return a + b
add = huey.task()(_add)
Task functions can be applied multiple times to a list (or iterable) of
parameters using the :py:meth:`~TaskWrapper.map` method:
.. code-block:: pycon
>>> @huey.task()
... def add(a, b):
... return a + b
...
>>> params = [(i, i ** 2) for i in range(10)]
>>> result_group = add.map(params)
>>> result_group.get(blocking=True)
[0, 2, 6, 12, 20, 30, 42, 56, 72, 90]
Retrieving Results by Task ID
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
If you have a task ID but not the original :py:class:`Result` handle (e.g.,
because you stored the ID in a database or session), you can retrieve the
result using :py:meth:`Huey.result`:
.. code-block:: python
# In a web handler, store the task ID when enqueuing:
result = process_upload(file_id)
session['task_id'] = result.id
# Later, in a status-checking endpoint:
task_id = session['task_id']
try:
value = huey.result(task_id, blocking=False, preserve=True)
except TaskException as exc:
return {'status': 'error', 'detail': exc.metadata['error']}
if value is None:
return {'status': 'pending'}
return {'status': 'complete', 'value': value}
Schedule Shorthand
^^^^^^^^^^^^^^^^^^
The :py:meth:`~TaskWrapper.schedule` method accepts a delay (int/float or
timedelta) or an ETA (datetime):
.. code-block:: python
# These are all equivalent:
add.schedule(args=(1, 2), delay=60)
add.schedule(args=(1, 2), delay=timedelta(seconds=60))
add.schedule(args=(1, 2), eta=datetime.now() + timedelta(seconds=60))
Iterating Results as They Complete
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
When working with a :py:class:`ResultGroup` (from :py:meth:`~TaskWrapper.map`
or :py:class:`group`), you can iterate over results as they become available
using :py:meth:`~ResultGroup.as_completed`:
.. code-block:: python
result_group = huey.enqueue(group([
fetch.s(url) for url in urls]))
for value in result_group.as_completed():
# Process each result as soon as it's ready, rather than waiting
# for all tasks to finish.
process(value)
Reserved Keyword Arguments
^^^^^^^^^^^^^^^^^^^^^^^^^^
When calling a task-decorated function or using :py:meth:`~TaskWrapper.s`, the
following keyword arguments are intercepted by Huey and will **not** be passed
to your task function:
* ``eta``
* ``delay``
* ``retries``
* ``retry_delay``,
* ``priority``
* ``expires``
* ``timeout``
This means you cannot use these names as keyword arguments to your task
function:
.. code-block:: python
@huey.task()
def bad_example(priority=None): # "priority" is reserved!
...
# This will set the task's priority to 10, NOT pass priority=10 to
# the function.
bad_example(priority=10)
If you need a parameter with one of these names, rename it or pass it
positionally.
Reading more
------------
That sums up the basic usage patterns of huey. Below are links for details on
other aspects of the APIs:
* :py:class:`Huey` - responsible for coordinating executable tasks and queue
backends
* :py:meth:`Huey.task` - decorator to indicate an executable task.
* :py:class:`Result` - handle for interacting with a task.
* :py:meth:`Huey.periodic_task` - decorator to indicate a task that executes at
periodic intervals.
* :py:func:`crontab` - define what intervals to execute a periodic command.
* For information about managing shared resources like database connections,
refer to the :ref:`shared resources ` document.
Also check out the :ref:`notes on running the consumer `.
---
## File: docs/imports.rst
.. _imports:
Understanding how tasks are imported
====================================
Behind-the-scenes when you decorate a function with :py:meth:`~Huey.task` or
:py:meth:`~Huey.periodic_task`, the function registers itself with an in-memory
registry. When a task function is called, a reference is put into the queue,
along with the arguments the function was called with, etc. The message is then
read by the consumer, and the task function is looked-up in the consumer's
registry. Because of the way this works, it is strongly recommended
that **all decorated functions be imported when the consumer starts up**.
If a task is not recognized, the consumer will raise a :py:class:`HueyException`.
The consumer is executed with a single required parameter, the import path to
a :py:class:`Huey` object. It will import the Huey instance along with
anything else in the module, so you must be sure **imports of your tasks
occur with the import of the Huey object**.
Suggested organization of code
------------------------------
Generally, I structure things like this, which makes it very easy to avoid
circular imports.
* ``config.py``, the module containing the :py:class:`Huey` object.
.. code-block:: python
# config.py
from huey import RedisHuey
huey = RedisHuey('testing')
* ``tasks.py``, the module containing any decorated functions. Imports the
``huey`` object from the ``config.py`` module:
.. code-block:: python
# tasks.py
from config import huey
@huey.task()
def add(a, b):
return a + b
* ``main.py`` / ``app.py``, the "main" module. Imports both the ``config.py``
module **and** the ``tasks.py`` module.
.. code-block:: python
# main.py
from config import huey # import the "huey" object.
from tasks import add # import any tasks / decorated functions
if __name__ == '__main__':
result = add(1, 2)
print('1 + 2 = %s' % result.get(blocking=True))
To run the consumer, point it at ``main.huey``, in this way, both the ``huey``
instance **and** the task functions are imported in a centralized location.
.. code-block:: shell
huey_consumer main.huey
---
## File: docs/index.rst
.. huey documentation master file, created by
sphinx-quickstart on Wed Nov 16 12:48:28 2011.
You can adapt this file completely to your liking, but it should at least
contain the root `toctree` directive.
huey
====
.. image:: https://media.charlesleifer.com/blog/photos/huey3-logo.png
*a lightweight alternative*.
huey is:
* a task queue
* written in python
* clean and simple API
huey has:
* support for redis (or valkey/redict), postgres, sqlite, file-system, or in-memory storage
* zero dependencies (``redis-py`` required to use redis-like brokers,
``psycopg`` for postgres).
* `example code `_.
* :ref:`django ` integration (native or via django.tasks) with
optional :ref:`admin integration ` for visibility and
management.
huey supports:
* multi-process, multi-thread or greenlet task execution models
* schedule tasks to execute at a given time, or after a given delay
* schedule recurring tasks, like a crontab
* automatically retry tasks that fail
* task prioritization
* task result storage
* task expiration
* task locking, rate-limits and timeouts
* task pipelines and chains
* groups (fan-out), chords (map / reduce)
.. image:: https://i.imgur.com/2EpRs.jpg
At a glance
-----------
:py:meth:`~Huey.task` and :py:meth:`~Huey.periodic_task` decorators turn
functions into tasks executed by the consumer:
.. code-block:: python
from huey import RedisHuey, crontab
huey = RedisHuey('my-app', host='redis.myapp.com')
@huey.task()
def add_numbers(a, b):
return a + b
@huey.task(retries=2, retry_delay=60)
def flaky_task(url):
# This task might fail, in which case it will be retried up to 2 times
# with a delay of 60s between retries.
return this_might_fail(url)
@huey.periodic_task(crontab(minute='0', hour='3'))
def nightly_backup():
sync_all_data()
Calling a ``task``-decorated function will enqueue the function call for
execution by the consumer. A special result handle is returned immediately,
which can be used to fetch the result once the task is finished:
.. code-block:: pycon
>>> from demo import add_numbers
>>> res = add_numbers(1, 2)
>>> res
>>> res()
3
Tasks can be scheduled to run in the future:
.. code-block:: pycon
>>> res = add_numbers.schedule((2, 3), delay=10) # Will be run in ~10s.
>>> res(blocking=True) # Will block until task finishes, in ~10s.
5
For much more, check out the :ref:`guide` or take a look at the `example code `_.
Running the consumer
^^^^^^^^^^^^^^^^^^^^
Run the consumer with four worker processes:
.. code-block:: shell
huey_consumer my_app.huey -k process -w 4
To run the consumer with a single worker thread (default):
.. code-block:: shell
huey_consumer my_app.huey
If your work-loads are mostly IO-bound, you can run the consumer with threads
or greenlets instead. Because greenlets are so lightweight, you can run quite a
few of them efficiently:
.. code-block:: shell
huey_consumer my_app.huey -k greenlet -w 32
For more information, see the :ref:`consuming-tasks` document.
Table of contents
-----------------
.. toctree::
:maxdepth: 2
installation
guide
consumer
deployment
imports
shared_resources
signals
api
contrib
troubleshooting
recipes
changes
Huey is named in honor of my cat
.. image:: https://m.charlesleifer.com/t/800x-/blog/photos/p1473037658.76.jpg?key=mD9_qMaKBAuGPi95KzXYqg
Indices and tables
==================
* :ref:`genindex`
* :ref:`modindex`
* :ref:`search`