Celery

---

Background

Celery is a distributed task queue that allows you to run tasks asynchronously in a distributed manner. It is often used to improve the performance and scalability of applications by offloading long-running or resource-intensive tasks to a separate worker process.

1. Simple Task Scheduling

from celery import Celery

app = Celery('tasks')

@app.task
def add(x, y):
    return x + y

add.delay(10, 20)

This schedules the add task to be executed asynchronously. The delay method returns a task object that you can use to check its status or retrieve its result.

2. Periodic Tasks

from celery.schedules import crontab

@app.task
def send_email():
    # Send an email

app.conf.beat_schedule = {
    'send-email': {
        'task': 'tasks.send_email',
        'schedule': crontab(hour=10, minute=0),
    }
}

This creates a periodic task that will run the send_email task every day at 10:00 AM.

3. Chaining Tasks

@app.task
def task1():
    return 'task1'

@app.task
def task2(x):
    return x + 'task2'

task2.s(task1.delay()).delay()

This chains the task1 and task2 tasks together. The task2 task will be executed with the result of the task1 task as input.

4. Retrying Tasks

@app.task(retry_backoff=True, retry_jitter=True)
def task():
    raise Exception('Error occurred')

task.delay()

This configures the task to be retried automatically if it fails. The retry_backoff and retry_jitter options control the delay between retries and the variation in retry delay, respectively.

5. Limiting Concurrency

@app.task(max_retries=5, soft_time_limit=300)
def task():
    # Run a long-running task

task.delay()

This limits the number of concurrent executions of the task to 5 and terminates the task if it takes longer than 5 minutes to complete.

6. Task Groups

from celery import group

tasks = group(add.s(x, y) for x in range(10) for y in range(10))
result = tasks.delay()

for task in result:
    print(task.get())

This creates a group of tasks that are executed in parallel. The result object can be used to retrieve the results of all the tasks in the group.

7. Task Chords

from celery import chord

add_tasks = [add.s(x, y) for x in range(10) for y in range(10)]
result = chord(add_tasks)(sum.s())

print(result.get())

This creates a chord that depends on the completion of the add_tasks tasks. The result object can be used to retrieve the final result of the chord, which is the sum of the results of the add_tasks.

8. Queues and Routing

app = Celery('tasks', broker='amqp://guest:guest@localhost:5672/', backend='redis://localhost:6379')

@app.task(queue='high-priority')
def task():
    # Run a high-priority task

@app.task(queue='low-priority')
def task():
    # Run a low-priority task

This creates two different queues, high-priority and low-priority. The tasks are routed to the appropriate queue based on their queue configuration.

9. Custom Task Classes

from celery import Task

class MyTask(Task):
    def run(self, x, y):
        return x + y

my_task = MyTask.delay(10, 20)

This creates a custom task class that inherits from the Celery.Task class. You can override the run method to customize the behavior of the task.

10. Custom Serializer

from celery import Serializer

class MySerializer(Serializer):
    def dumps(self, task_data):
        # Custom encoding logic

    def loads(self, task_data):
        # Custom decoding logic

app = Celery('tasks', serializer='tasks.MySerializer')

This creates a custom serializer that serializes and deserializes task data. You can use this to customize the way task data is stored and transmitted.