pip freeze --local

def publish(self, queue: str, message: dict | str):
    logger.info(f'Publishing message to queue: {queue}')
    self.channel.queue_declare(queue=queue, durable=True)
    if isinstance(message, dict):
        message = json.dumps(message)

    self.channel.basic_publish(
        exchange=self.exchange,
        routing_key=queue,
        body=message,
        properties=pika.BasicProperties(delivery_mode=2)
    )
    logger.info('Message successfully published.')

from core.celery_app import celery_app

@celery_app.task(bind=True, retry=5, default_retry_delay=60)
def consume_message(self, event_body):
    <put-your-logic-here>

import django
from celery import Celery
from celery.fixups.django import DjangoFixup
from django.apps import apps
from django.conf import settings
from kombu.common import Broadcast, Queue

from core.settings import project_settings


celery_app = Celery('core')
celery_app.config_from_object('django.conf:settings')
DjangoFixup(celery_app).install()

if not settings.configured:
    settings.configure(**project_settings.data)
    django.setup()

celery_app.autodiscover_tasks(lambda: [n.name for n in apps.get_app_configs()])
celery_app.conf.task_queues = (Queue('name-of-your-queue'))
celery_app.conf.task_routes = {
    'tasks.consume_message': {
        'queue': 'name-of-your-queue',
    },
}
celery_app.conf.result_extended = True

class RabbitMQProducer:

    def __init__(
        self,
        host: str = settings.BROKER_HOST,
        port: int = settings.BROKER_PORT,
        heartbeat: int = settings.BROKER_HEARTBEAT,
        blocked_connection_timeout: int = settings.BLOCKED_CONNECTION_TIMEOUT,
        password: str = settings.BROKER_PASSWORD,
        username: str = settings.BROKER_USERNAME,
        vhost: str = settings.BROKER_VHOST,
        exchange: str = settings.BROKER_EXCHANGE
    ):
        self.exchange = exchange
        self.credentials = pika.PlainCredentials(username, password)
        self.connection_params = pika.ConnectionParameters(
            host=host,
            port=port,
            heartbeat=heartbeat,
            blocked_connection_timeout=blocked_connection_timeout,
            credentials=self.credentials,
            virtual_host=vhost
        )
        self.connection = pika.BlockingConnection(self.connection_params)
        self.channel = self.connection.channel()

    def publish(self, queue: str, message: dict | str):
        logger.info(f'Publishing message to queue: {queue}')
        self.channel.queue_declare(queue=queue, durable=True)
        if isinstance(message, dict):
            message = json.dumps(message)

        message_with_celery_body = {
            "args": [message],
            "kwargs": {}
        }
        self.channel.basic_publish(
            exchange=self.exchange,
            routing_key=queue,
            body=json.dumps(message_with_celery_body),
            properties=pika.BasicProperties(
                delivery_mode=2,
                headers={'id': str(uuid.uuid4()), 'task': 'tasks.consume_message'},
                content_type='application/json'
            )
        )
        logger.info('Message successfully published.')