celery-once
celery-once copied to clipboard
self.retry doesn't retry the task on unittests
I have a celery task which base class is QueueOnce, the task is locked with order_id argument:
@app.task(
bind=True,
soft_time_limit=settings.SOFT_TIME_LIMIT,
base=QueueOnce,
once={'keys': ['order_id'],
'graceful': False}
)
def some_celery_task_invoke(self, order_id, **kwargs):
task = SomeTask()
retry, res = task.run()
...
if retry:
try:
retry_parameters = {
'kwargs': res,
'countdown': delay_seconds,
'max_retries': max_retries,
'soft_time_limit': time_limit,
}
self.retry(**retry_parameters)
except exceptions.MaxRetriesExceededError:
logger.info('Some logging')
This task works well and retries are working as well when I am testing it on dev environment. But, when I run a unittest to check if retries are working, task call self.retry(**retry_parameters) - it doesn't initiate some_celery_task_invoke. What could be the reason?