Via RabbitMQ CLI

1. Delete the default user: rabbitmqctl delete_user guest.

2. Create a new user: rabbitmqctl add_user user_name user_password.

3. Set permissions to the new user: rabblitmqcctl set_permissions -p / user_name ".*" ".*" ".*".

4. Modify URL for Celery: amqp://user_name:user_password@rabbitmq:5672/.

RabbitMQ management

1. Restrict access to the 15672port using NF Tables.

To use Kafka as a broker for Celery, you need to configure it explicitly because Kafka is not natively supported out of the box like Redis or RabbitMQ. Celery supports Kafka through the kafka-python library.

1. Install Kafka and Kafka-Python: pip install kafka-python

2. Configure Celery to Use Kafka: app = Celery('myproject', broker='kafka://localhost:9092') at celery.py

Example Cnfiguration with custom options:

from celery import Celery

app = Celery('myproject', broker='kafka://localhost:9092')

# Additional configurations (e.g., setting default topic)
app.conf.update(
    broker_transport_options={
        'topic': 'my_celery_topic',  # Set your desired topic
        'acks': 'all',  # Kafka acknowledgments
    }
)

1. Install Flower via PIP: pip install flower.

2. Start flower: celery -A your_project_name flower.

3. Access flower dashboard: http://localhost:5555.

Django-celery-beat Installation and Set up

1. Install: pip install djang-celery-beat

2. Add Django-celery-beat to the Installed apps django_celery_beat.

3. Run migrations.

4. Update Celery configuration:

   Ensure your Celery configuration in celery.py or settings.py includes the django_celery_beat.schedulers:DatabaseScheduler as your task scheduler:

   from celery import Celery
   
   app = Celery('your_project_name')
   
   # Your other Celery configuration here
   
   app.conf.beat_scheduler = 'django_celery_beat.schedulers:DatabaseScheduler'

5. Start Using the Django Admin

• Port: Specify a different port if you don’t want to use the default port 5555:

  celery -A your_project_name flower --port=5566

• Basic Authentication: Set up basic authentication for accessing flower:

  celery -A your_project_name flower --basic_auth=username:password

• Persistent data:

  celery -A your_project_name flower --persistent=True

• URL prefix:

  celery -A your_project_name flower --url_prefix=flower

• Logging:

Shared tasks are a way to define reusable tasks that can be registered with multiple Celery applications. This is useful in a Django project where you might have multiple Django apps with their own tasks, but want to avoid tight coupling between them.

Key Features of Shared Tasks:

• Independence: Shared tasks are defined without being bound to a specific Celery instance or app. This makes them reusable across different Celery apps and modules.

• Flexibility: By using shared tasks, you can define a task once and use it in multiple places. This is helpful for maintaining DRY (Don't Repeat Yourself) principles in your code.

• Registration: A shared task is registered with the Celery app that imports it. When the task is executed, it is registered with the current app automatically.

from __future__ import absolute_import, unicode_literals
from celery import shared_task

@shared_task
def add(x: int, y: int) -> int:
	return x + y

• A regular task is the most basic form of a Celery task. It's a Python function decorated with the @task decorator from celery and can be executed asynchronously by the worker processes.

• These tasks can be simple functions that perform a specific operation like sending an email, performing a calculation, or interacting with an external API.

from celery import Celery

app = Celery('my_app')

@app.task
def send_email(recipient, subject, message):
    # Logic to send email
    pass

• Periodic tasks are tasks that run at regular intervals. They are defined similarly to regular tasks but are scheduled using Celery's beat scheduler or by integrating with a periodic task scheduler like celery-beat.

• These tasks are useful for recurring operations, like cleaning up old data, sending periodic notifications, or updating cache.

from celery.schedules import crontab

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # Executes every Monday morning at 7:30 a.m.
    sender.add_periodic_task(
        crontab(hour=7, minute=30, day_of_week=1),
        my_periodic_task.s(),
    )

@app.task
def my_periodic_task():
    # Task logic here
    pass

• Chords: A chord is a task that only executes after a group of other tasks has completed. It's useful for aggregating results or running follow-up tasks.

• Chains: A chain is a sequence of tasks that must be executed one after the other. Each task passes its result to the next task in the chain.

• Groups: A group is a collection of tasks that can be executed in parallel. Groups are used to perform multiple tasks simultaneously and collect their results.

from celery import group, chain, chord

# Example of a group
group_tasks = group(task1.s(), task2.s(), task3.s())
result = group_tasks.apply_async()

# Example of a chain
chain_tasks = chain(task1.s(), task2.s(), task3.s())
result = chain_tasks.apply_async()

# Example of a chord
chord_tasks = chord((task1.s(), task2.s()), task3.s())
result = chord_tasks.apply_async()

Callbacks are tasks that are executed after another task finishes. Celery allows you to specify a callback task that runs after the main task is completed.

Callback tasks are extremely useful when you need to run a sequence of tasks where each task depends on the result of the previous one.

You can trigger a callback task by passing the link argument when calling a Celery task. The task specified as the callback will be executed after the original task is completed.

# tasks.py
from celery import shared_task

@shared_task
def first_task(x, y):
    return x + y

@shared_task
def callback_task(result):
    print(f'The result of the first task is {result}')

# Somewhere in your code
result = first_task.apply_async((4, 6), link=callback_task.s())

@app.task
def task_with_callback():
    # Main task logic
    pass

@app.task
def on_success(result):
    # Logic to execute on success
    pass

task_with_callback.apply_async(callback=on_success.s())

The .delay() method is a shortcut to call a task asynchronously. It sends the task to the Celery worker for execution. This method is simple and straightforward, and it doesn't require any additional configuration.

# Assume you have a Celery task defined as follows:
@app.task
def add(x, y):
    return x + y

# Call the task asynchronously using delay()
result = add.delay(4, 6)

• It is commonly used for simple use cases where no additional task options are needed (like setting a countdown or a specific queue).

• Returns an AsyncResult object, which can be used to check the status of the task or get the result once it’s completed.

The .apply_async() method provides more flexibility and allows you to specify additional options like countdown, retries, task routing, etc. It's a more powerful way to call tasks compared to .delay().

# Call the task asynchronously with additional options using apply_async()
result = add.apply_async((4, 6), countdown=10, queue='priority_queue')

Key Points:

• .apply_async() takes two main arguments: a tuple of args and a dictionary of kwargs for task arguments.

• It allows specifying additional options like:
  • countdown: Number of seconds to delay execution.
  • eta: Exact date and time to execute the task.
  • expires: Date and time (or seconds) after which the task should not be executed.
  • retry: Whether to retry the task if it fails.
  • queue: Specify the name of the queue to send the task to.
  • priority: Set the priority of the task (usually between 0 and 255).

• Returns an AsyncResult object for monitoring the task.

The .run() method calls the task function directly without sending it to the Celery worker. This is a plain Python function call.

# Call the task directly
result = add.run(4, 6)

Task signatures are a way to create a task call that can be reused, scheduled, or combined with other tasks (like in chains, groups, or chords).

# Create a task signature
signature = add.s(4, 6)

# Call the task asynchronously using the signature
result = signature.apply_async()

# Or using delay on signature
result = signature.delay()

Key Points:

• Signatures provide a flexible way to manage task execution, especially when working with complex workflows.

• They can be passed around, stored, and executed later.

⚠️NOTE: Do know forget to initialize celery beats correctly if you need task scheduled in via Django Admin panel run.

Django-Celery-beat is a Django extension that allows you to manage periodic tasks in a Django application using the Django admin interface. It integrates Celery with Django by providing a way to schedule tasks periodically, leveraging Django's ORM and admin functionalities.

from celery.exceptions import MaxRetriesExceededError
from .email import send_verification_code_email


@shared_task(bind=True, name="send_verification_code", max_retries=3, default_retry_delay=30)
def send_verification_code_task(self, email: str, code: int | str):
    try:
        send_verification_code_email(email, code)
    except Exception as exc:
        try:
            raise self.retry(exc=exc)
        except MaxRetriesExceededError:
            print(f"Failed to send the verification code after ({self.max_retries})")

1. Create a new Django app (python3 manage.py startapp emailSender)

2. Create a new form at forms.py

3. create view.py

4. create the send_email method in forms.py

5. create and configure the tasks.py

Configure email

1. Create emial.py

1. Install Gunicorn: pip install gunicorn.

2. Configure Django for production:
   1. Update settings.py with production settings.
   2. Collect static files by running python3 manage.py collectstatics.

3. Run Gunicorn: gunicorn myproject.wsgi:application --bind 0.0.0.0:8000 --workers 3. Or you might want to create a Gunicorn config file:

   [Unit]
   Description=Gunicorn instance to serve Django application
   After=network.target
   
   [Service]
   User=youruser
   Group=yourgroup
   WorkingDirectory=/path/to/your/project
   ExecStart=/path/to/your/virtualenv/bin/gunicorn --workers 3 --bind unix:/path/to/your/project.sock myproject.wsgi:application
   
   [Install]
   WantedBy=multi-user.target

1. Install Celery pip install celery.

2. Configure Celery in Django:

   CELERY_BROKER_URL = 'redis://localhost:6379/0'  # Example using Redis
   CELERY_ACCEPT_CONTENT = ['json']
   CELERY_TASK_SERIALIZER = 'json'
   CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

   b. Create a celery.py file in the Project root directory.

   from __future__ import absolute_import, unicode_literals
   import os
   from celery import Celery
   
   # Set the default Django settings module for the 'celery' program.
   os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
   
   app = Celery('myproject')
   
   # Using a string here means the worker doesn't have to serialize
   # the configuration object to child processes.
   # - namespace='CELERY' means all celery-related configuration keys
   #   should have a `CELERY_` prefix.
   app.config_from_object('django.conf:settings', namespace='CELERY')
   
   # Load task modules from all registered Django app configs.
   app.autodiscover_tasks()
   
   @app.task(bind=True)
   def debug_task(self):
       print(f'Request: {self.request!r}')

   c. Configure __init__.pyfile in the Project root directory.

   from __future__ import absolute_import, unicode_literals
   
   # This will make sure the app is always imported when
   # Django starts so that shared_task will use this app.
   from .celery import app as celery_app
   
   __all__ = ('celery_app',)

3. Run a Celery worker: celery -A myproject worker -l info. Similar to Gunicorn you might want to create a config file for Celery:

   [Unit]
   Description=Celery Service
   After=network.target
   
   [Service]
   Type=forking
   User=youruser
   Group=yourgroup
   EnvironmentFile=/etc/default/celeryd
   WorkingDirectory=/path/to/your/project
   ExecStart=/path/to/your/virtualenv/bin/celery multi start worker -A myproject --loglevel=info --logfile=/var/log/celery/worker.log --pidfile=/var/run/celery/worker.pid
   ExecStop=/path/to/your/virtualenv/bin/celery multi stopwait worker --pidfile=/var/run/celery/worker.pid
   ExecReload=/path/to/your/virtualenv/bin/celery multi restart worker -A myproject --loglevel=info --logfile=/var/log/celery/worker.log --pidfile=/var/run/celery/worker.pid
   
   [Install]
   WantedBy=multi-user.target

1. install Django-Celery-Beat: pip install django-celery-beat

2. Add to INSTALLED_APPS: django_celery_beat.

3. Run migrations: python3 manage.py migrate.

4. Run Celery beat worker: celery -A myproject beat -l info --scheduler django_celery_beat.schedulers:DatabaseSchedule.

1. Install Flower: pip install flower .

2. Run flower: celery -A myproject flower.