Run and monitor celery tasks
Project description
Summary
Run, monitor and log celery tasks.
Installation and setup
Declare tasks using celery task or cubicweb-celery cwtasks.
On worker side, install cw-celerytask-helpers.
celeryconfig.py example:
CELERY_BROKER_URL = 'redis://localhost:6379/0' CELERY_RESULT_BACKEND = BROKER_URL CUBICWEB_CELERYTASK_REDIS_URL = CELERY_BROKER_URL CELERY_IMPORTS = ('cw_celerytask_helpers.helpers', 'module.containing.tasks')
In this configuration example, the cw_celerytask_helpers in CELERY_IMPORTS is required to have logging data (in the task) sent back to the Cubicweb instance via Redis. The CUBICWEB_CELERYTASK_REDIS_URL is the Redis endpoint used for this logging handling mechanism.
Start a worker:
# running cubicweb tasks (celeryconfig.py will be imported from your instance config directory) celery -A cubicweb_celery -i <CW_INSTANCE_NAME> worker -l info # running pure celery tasks celery worker -l info
Task state synchronization requires to run the celery-monitor command:
cubicweb-ctl celery-monitor <instance-name>
Ensure to have the celeryconfig.py loaded for both cubicweb instance and celery worker, enforce by settings with CELERY_CONFIG_MODULE environment variable (it must be an importable python module).
Running tasks
Create a task:
from celery import current_app as app
from celery.utils.log import get_task_logger
logger = get_task_logger(__name__)
@app.task(name='hi_there')
def my_task(arg, kw=0):
logger.info('HI %s %s!', arg, kw)
return 42
Run a task:
from cubicweb_celerytask.entities import start_async_task
cwtask = start_async_task(cnx, 'hi_there', 'THERE', kw=42)
cnx.commit()
start_async_task() accept task names, task objects or task signatures: http://docs.celeryproject.org/en/latest/userguide/canvas.html#signatures
For instance, to start the above task in a dedicated queue named myqueue:
import celery
start_async_task(cnx, celery.signature('hi_there', args=('THERE',),
kwargs={'kw': 42}, queue='myqueue'))
Testing task based application
In CubicWeb test mode, tasks don’t run automatically, use cubicweb_celerytask.entities.get_tasks() to introspect them and cubicweb_celerytask.entities.run_all_tasks() to run them.
Also, CELERY_ALWAYS_EAGER and CELERY_EAGER_PROPAGATES_EXCEPTIONS are set to True by default.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Hashes for cubicweb-celerytask-2.0.1.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 1ffd3504d354d4ebefa30631f0c72c17f620ba51c4d6c6ba42be52a229650b5e |
|
MD5 | f8a935c6649486bd0d29a4f7d767f701 |
|
BLAKE2b-256 | 016b9f412c2a8f978332b66318df1986194ef29ae7dbf08233b94919a586e849 |