Problem with redis/celery, why do some tasks triggers an Errno 111 but others don't?

I am currently building a web app with django. I am using redis and celery to send emails for various purposes. I have a background worker checking my db for mails to send every 5s (very short for testing purposes),if it finds mails flagged as unsent, then it sends them, this part works as intended. However when I call the same method directly from one of my views (to send an email for account creation), I have an Errno 111.

Here's my celery.py file:

import os
from celery import Celery
from celery.signals import after_setup_logger
import logging.config

from inscriptionsIff import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'inscriptionsIff.settings')

app = Celery('inscriptionsIff')
app.config_from_object('django.conf:settings', namespace='CELERY')


@after_setup_logger.connect
def setup_loggers(logger, *args, **kwargs):
    logging.config.dictConfig(settings.LOGGING)


app.conf.beat_schedule = {
    'send-emails-every-x-seconds': {
        'task': 'inscriptionsIff.tasks.process_and_send_emails',
        'schedule': 5,  # Replace 'x' with the number of seconds
    },
}

app.autodiscover_tasks()

Here's my tasks.py file:

from celery import shared_task
from django.core.mail import send_mail
from .models import Submission
from django.urls import reverse
from django.contrib.sites.shortcuts import get_current_site
import hashlib
from django.conf import settings


@shared_task
def send_email_task(email, subject, message):
    send_mail(
        subject,
        message,
        'yourgmailaddress@gmail.com',
        [email],
        fail_silently=False,
    )


def hash_input(input_to_hash):
    # Use Django's SECRET_KEY to add a layer of unpredictability to the hash
    salted_id = str(input_to_hash) + settings.SECRET_KEY
    return hashlib.sha256(salted_id.encode()).hexdigest()


@shared_task
def process_and_send_emails():
    # Get the current site domain for constructing absolute URLs
    # current_site = get_current_site(None)

    for record in Submission.objects.filter(mail_sent=False).only('email', 'id').iterator():
        # Generate the hashed ID
        submission_hash = hash_input(record.id)

        # Build the absolute confirmation URL
        path = reverse('confirm-reception', kwargs={'submission_hash': submission_hash})
        confirm_url = f'http://127.0.0.1:8080{path}'

        # Construct the email message with the confirmation link
        message = f'TEST : \n{confirm_url}'

        # Send the email
        send_email_task.delay(
            email=record.email,
            subject='Journée de formation SNCF & Moi',
            message=message,
        )

        # Mark the email as sent
        record.mail_sent = True
        record.save(update_fields=['mail_sent'])

And finally,here's the view that creates my user:

@csrf_exempt
def create_super_user(request):
    if request.method == 'POST':
        email = request.POST.get('email')
        password = request.POST.get('password')

        if not email or not password:
            return HttpResponse('Email and password are required.', status=400)

        if CustomUser.objects.filter(email=email).exists():
            return HttpResponse('A user with this email already exists.', status=400)

        # Create the Account instance first
        account = Account.objects.create()

        
        super_user = CustomUser.objects.create_user(
            email=email,
            password=password,  
            is_superuser=True,
            is_active=True,  
            account_id=account.id,  
        )
        confirmation_hash = hash_input(email)
        path = reverse('verify_email', kwargs={'confirmation_hash': confirmation_hash})
        confirm_url = f'http://127.0.0.1:8080{path}'
        send_email_task.delay(email=email, message=confirm_url, subject='Test')
        

        

        return HttpResponse('Super user and account created successfully.')
    else:
        return render(request, 'account_creation.html')

After testing, I'm sure that the send_email_task.delay within the view triggers the error, but don't understand why, as the background mail task works perfectly fine.

My broker url is defined like so:

CELERY_BROKER_URL = 'redis://localhost:6379/0'

Here's the complete error log :

Environment:


Request Method: POST
Request URL: http://127.0.0.1:8080/create_super_user

Django Version: 5.0.2
Python Version: 3.10.12
Installed Applications:
['django.contrib.admin',
 'django.contrib.auth',
 'django.contrib.contenttypes',
 'django.contrib.sessions',
 'django.contrib.messages',
 'django.contrib.staticfiles',
 'inscriptionsIff']
Installed Middleware:
['django.middleware.security.SecurityMiddleware',
 'django.contrib.sessions.middleware.SessionMiddleware',
 'django.middleware.common.CommonMiddleware',
 'django.middleware.csrf.CsrfViewMiddleware',
 'django.contrib.auth.middleware.AuthenticationMiddleware',
 'django.contrib.messages.middleware.MessageMiddleware',
 'django.middleware.clickjacking.XFrameOptionsMiddleware']



Traceback (most recent call last):
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/utils/functional.py", line 32, in __call__
    return self.__value__

During handling of the above exception ('ChannelPromise' object has no attribute '__value__'), another exception occurred:
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/connection.py", line 472, in _reraise_as_library_errors
    yield
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/connection.py", line 459, in _ensure_connection
    return retry_over_time(
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/utils/functional.py", line 318, in retry_over_time
    return fun(*args, **kwargs)
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/connection.py", line 934, in _connection_factory
    self._connection = self._establish_connection()
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/connection.py", line 860, in _establish_connection
    conn = self.transport.establish_connection()
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/transport/pyamqp.py", line 203, in establish_connection
    conn.connect()
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/amqp/connection.py", line 324, in connect
    self.transport.connect()
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/amqp/transport.py", line 129, in connect
    self._connect(self.host, self.port, self.connect_timeout)
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/amqp/transport.py", line 184, in _connect
    self.sock.connect(sa)

The above exception ([Errno 111] Connection refused) was the direct cause of the following exception:
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/django/core/handlers/exception.py", line 55, in inner
    response = get_response(request)
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/django/core/handlers/base.py", line 197, in _get_response
    response = wrapped_callback(request, *callback_args, **callback_kwargs)
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/django/views/decorators/csrf.py", line 65, in _view_wrapper
    return view_func(request, *args, **kwargs)
  File "/home/franzozich/PycharmProjects/inscriptionsIff/inscriptionsIff/views.py", line 147, in create_super_user
    send_email_task.delay(email=email, message=confirm_url, subject='Test')
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/celery/app/task.py", line 444, in delay
    return self.apply_async(args, kwargs)
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/celery/app/task.py", line 594, in apply_async
    return app.send_task(
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/celery/app/base.py", line 799, in send_task
    amqp.send_task_message(P, name, message, **options)
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/celery/app/amqp.py", line 518, in send_task_message
    ret = producer.publish(
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/messaging.py", line 186, in publish
    return _publish(
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/connection.py", line 556, in _ensured
    return fun(*args, **kwargs)
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/messaging.py", line 195, in _publish
    channel = self.channel
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/messaging.py", line 218, in _get_channel
    channel = self._channel = channel()
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/utils/functional.py", line 34, in __call__
    value = self.__value__ = self.__contract__()
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/messaging.py", line 234, in <lambda>
    channel = ChannelPromise(lambda: connection.default_channel)
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/connection.py", line 953, in default_channel
    self._ensure_connection(**conn_opts)
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/connection.py", line 458, in _ensure_connection
    with ctx():
  File "/usr/lib/python3.10/contextlib.py", line 153, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/home/franzozich/PycharmProjects/inscriptionsIff/.venv/lib/python3.10/site-packages/kombu/connection.py", line 476, in _reraise_as_library_errors
    raise ConnectionError(str(exc)) from exc

Exception Type: OperationalError at /create_super_user
Exception Value: [Errno 111] Connection refused

I'm starting my app with the following shell script :

#!/bin/bash

# Function to stop all background processes
cleanup() {
    echo "Stopping Django development server..."
    kill $DJANGO_PID

    echo "Stopping Celery worker..."
    kill $CELERY_WORKER_PID

    echo "Stopping Celery Beat..."
    kill $CELERY_BEAT_PID

    echo "Stopping Redis server..."
    redis-cli shutdown
}

# Trap SIGINT (Ctrl+C) and SIGTERM signals to cleanup before exiting
trap cleanup SIGINT SIGTERM

# Start Redis server in the background
echo "Starting Redis server..."
redis-server &

# Start Django development server in the background and save its PID
echo "Starting Django development server..."
python manage.py runserver 8080 &
DJANGO_PID=$!

# Start Celery worker in the background and save its PID
echo "Starting Celery worker..."
celery -A inscriptionsIff worker --loglevel=info &
CELERY_WORKER_PID=$!

# Start Celery Beat in the background and save its PID
echo "Starting Celery Beat..."
celery -A inscriptionsIff beat --loglevel=info &
CELERY_BEAT_PID=$!

# Wait for all background processes to finish
wait $DJANGO_PID $CELERY_WORKER_PID $CELERY_BEAT_PID

This is a complete mystery, I've checked that the redis server answers properly once the app is started, as the background mail sending service works properly, I have no idea why calling this task from one of my views triggers this sort of error. Does anyone have any ideas ?

Back to Top