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
Built Distribution
Hashes for cubicweb-celerytask-0.7.5.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | f5d806f32e8681f748fd583a61c819ee2ba7b9c52245a97d486b019e88229d64 |
|
MD5 | 676c8fe3c2babede23ac18f881aad5b5 |
|
BLAKE2b-256 | 28ffc10fb90fe4928e27d723f1bf905a15afe5e3353526dea4aa1c2505fc9a8b |
Hashes for cubicweb_celerytask-0.7.5-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | ff7f57da9cf36c81827ffa81e4c35b5e65a6c23cd0dd219f5cb50574d39bc8ea |
|
MD5 | 8bef75273e816c13ef5666eea0443596 |
|
BLAKE2b-256 | 220387351853b6ea534eaa29e51b69529c5b505da926f736ac95b47267066956 |