run and retry Celery tasks synchronously

celery_sync.py

"""
useful for local development and unit tests when tasks are run syncronously
"""

from your_app.settings import TESTING
from celery import Task
from celery.canvas import Signature

def run_task(task_signature: Signature):
    if TESTING:
        task_signature()
    else:
        task_signature.delay()


def retry_task(task: Task, task_kwargs,  retry_kwargs):
    if TESTING:
        task.request.retries += 1
        task(**task_kwargs)
    else:
        task.retry(**retry_kwargs)