darq 0.7.1

A small wrapper around arq

Async task manager with Celery-like features. Fork of arq.

Features

• Celery-like @task decorator, adds .delay() to enqueue job

• Graceful shutdown: waits until running tasks are finished

Quick start

# some_project/darq_app.py
import asyncio
import darq

darq = darq.Darq(redis_settings=darq.RedisSettings(host='redis'))


@darq.task
async def add_to_42(a: int) -> int:
    return 42 + a


async def main():
    # Before adding tasks to queue we should connect darq instance to redis
    await darq.connect()

    # Direct call job as function:
    result = await add_to_42(5)  # result == 47

    # Celery-like add task to queue:
    await add_to_42.delay(a=5)

    await darq.disconnect()


if __name__ == '__main__':
    asyncio.run(main())

And start worker:

python3 -m darq.cli some_project.darq_app.darq

Worker output:

15:24:42: Starting worker for 1 functions: some_project.darq_app.add_to_42
15:24:42: redis_version=5.0.7 mem_usage=834.87K clients_connected=1 db_keys=2
15:25:08:   0.22s → 1315f27608e9408392bf5d3310bca38c:darq_app.add_to_42(a=5)
15:25:08:   0.00s ← 1315f27608e9408392bf5d3310bca38c:darq_app.add_to_42 ● 47

Changelog

0.7.1 (2020-05-25)

• .apply_async: Make args and kwargs arguments optional

0.7.0 (2020-05-25)

• Fork arq to project and merge it with darq (It was easier to rewrite arq than to write a wrapper)

• Breaking change: Remove “magic” params from .delay. For enqueue job with special params added .apply_async.

• Add watch-mode to CLI.

• Fix: Now worker will not run cronjob if it’s functions queue not match with worker’s

0.6.0 (2020-03-08)

• Breaking change: Changed Darq constructor from single config param to separate params.

• arq_function.coroutine now has .delay method.

0.5.0 (2020-03-03)

• Add on_job_prepublish(metadata, arq_function, args, kwargs) callback. metadata is mutable dict, which will be available at ctx['metadata'].

0.4.0 (2020-03-03)

• Add default_job_expires param to Darq (if the job still hasn’t started after this duration, do not run it). Default - 1 day

• Add expires param to @task (if set - overwrites default_job_expires)

0.3.1 (2020-03-02)

• Rewrite warm shutdown: now during warm shutdown cron is disabled, on second signal the warm shutdown will be canceled

0.3.0 (2020-02-27)

• Breaking change: on_job_prerun and on_job_postrun now accepts arq.worker.Function instead of the original function (it can still be accessed at arq_function.coroutine)

0.2.1 (2020-02-26)

• Fix add_cron_jobs method. Tests added.

0.2.0 (2020-02-26)

• Add on_job_prerun(ctx, function, args, kwargs) and on_job_postrun(ctx, function, args, kwargs, result) callbacks.

0.1.0 (2020-02-26)

• Breaking change: Jobs no longer explicitly get JobCtx as the first argument, as in 99.9% cases it doesn’t need it. In future release will be possible to optionally pass JobCtx in some way.

• Breaking change: All cron jobs should be wrapped in @task decorator

• Directly pass functions to arq.Worker, not names.

0.0.3 (2020-02-25)

• .delay() now returns arq_redis.enqueue_job result (Optional[Job])

• Add py.typed file

• Fixed add_cron_jobs typing

0.0.2 (2020-02-24)

• Add add_cron_jobs method

0.0.1 (2020-02-21)

First release