Django-Celery перестал работать после переключения ветвей

Я использовал cookiecutter-django для создания своего Django проекта с celery=yes и docker=yes. Я потратил некоторое время на изменение конфигурации/структуры Celery в соответствии с руководством HackSoftware Django Guidelines для Celery. Мне потребовалось время, чтобы во всем разобраться, но в конце концов я добился того, что все заработало с примером задачи, которая просто записывает в журнал несколько строк.

Это было сделано на ветви feature. Итак, как только я убедился, что все в порядке, я зафиксировал и продвинул свои изменения и объединил PR с основной веткой. Вернувшись в терминал, я переключился на Main и вытащил изменения. В этот момент не должно быть различий между моей веткой main и моей веткой configure_celery. Однако, когда я поднял все контейнеры, моя задача-образец больше не выполнялась.

Я потратил много часов, пробуя различные вещи, чтобы заставить его работать безрезультатно. Я снес контейнеры, обрезал все с помощью docker system prune -a, использовал Docker Desktop, чтобы вручную очистить все тома, попробовал другую конфигурацию Redis db, вернулся к моей ветке configure_celery и ничего не сработало.

Я вижу, что рабочий успешно обнаруживает мою задачу:

Все контейнеры выглядят счастливыми, и их службы запущены:

brew_dash_local_celeryworker   "/entrypoint /start-…"   celeryworker        running
brew_dash_local_django         "/entrypoint /start"     django              running             0.0.0.0:8000->8000/tcp
brew_dash_local_flower         "/entrypoint /start-…"   flower              running             0.0.0.0:5555->5555/tcp
brew_dash_local_mailhog        "MailHog"                mailhog             running             0.0.0.0:8025->8025/tcp
brew_dash_local_postgres       "docker-entrypoint.s…"   postgres            running             0.0.0.0:5432->5432/tcp
brew_dash_local_redis          "docker-entrypoint.s…"   redis               running             6379/tcp

Мой конфиг находится по адресу project/brew_dash/tasks/apps.py:

import os

from celery import Celery
from django.apps import AppConfig

# set the default Django settings module for the 'celery' program.
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.local")

app = Celery("brew_dash")


# TODO add a periodic task script like here: https://github.com/HackSoftware/Django-Styleguide#periodic-tasks. This
#  will be really handy for deployments


class TasksConfig(AppConfig):
    """Most of the config is based on the HackSoftware Django-Styleguidelines:
    https://github.com/HackSoftware/Django-Styleguide#celery"""

    name = "brew_dash.tasks"
    verbose_name = "Celery Config"

    def ready(self):
        # 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.
        # Each app that will have tasks need to have its tasks file registered in AppConfig
        # Per: https://cheat.readthedocs.io/en/latest/django/celery.html
        app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    from celery.utils.log import base_logger

    base_logger = base_logger

    base_logger.debug("debug message")
    base_logger.info("info message")
    base_logger.warning("warning message")
    base_logger.error("error message")
    base_logger.critical("critical message")

    print("Request: {0!r}".format(self.request))

    return 42

project/config/init.py

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from brew_dash.tasks.apps import app as celery_app

__all__ = ("celery_app",)

project/integrations/tasks.py:

from celery import shared_task
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)


@shared_task
def sample_task():
    logger.info("***********************************************************")
    logger.info("***********************************************************")
    logger.info("The sample task just ran ****************")
    logger.info("***********************************************************")

project/brew_dash/integrations/apps.py:

class IntegrationsConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "brew_dash.integrations"

    def ready(self):
        import brew_dash.integrations.tasks  # NOQA

Я создал интервал, который запускается каждые 15 секунд в приложении Django admin, и привязал его к периодической задаче sample_task, которую обнаруживает рабочий. Все это работало до того, как я выложил свои изменения, объединил и попробовал запустить на основной ветке. Я проделал все известные мне способы устранения неполадок, так что я ищу совета. Спасибо!

Вернуться на верх