Сообщение Slack не отправляется, когда python-код подключен к celery

Я создал проект django для загрузки статей. В нем я создал функциональность, в которой, если пользователь не загрузил статью до 11:00 утра, в slack будет отправлено сообщение о необходимости загрузить статью.

Проблема в том, что если я не добавляю задачу в celery и выполняю ее самостоятельно, то она работает. Но в сочетании с celery - нет.

Рабочий сельдерея выдает сообщение :

[2024-02-06 20:49:49,015: INFO/MainProcess] Task account.tasks.check_articles_and_send_slack_message[fdc4f121-f123-4697-85c2-ce2bb57e6004] received
[2024-02-06 20:49:49,017: INFO/MainProcess] Task account.tasks.check_articles_and_send_slack_message[fdc4f121-f123-4697-85c2-ce2bb57e6004] succeeded in 0.**strong text**

но все еще не отправлено ни одного сообщения о слабине.

Это файл tasks.py, который находится в приложении "Счета":

from celery import shared_task
from django.utils import timezone
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from writer.models import Article

@shared_task
def check_articles_and_send_slack_message():
    articles = Article.objects.filter(date_posted__date=timezone.now().date())
    for article in articles:
        if article.date_posted.time() < timezone.time(11, 0, 0):
            send_slack_message(article.user)
    
def send_slack_message(user):
    client = WebClient(token='dummy-token')
    try:
        response = client.chat_postMessage(
            channel='#qa',
            text=f"Hi @{user.first_name}, please upload your article before 11:00 a.m. today."
        )
        print("Slack message sent successfully")
    except SlackApiError as e:
        print(f"Error sending Slack message: {e.response['error']}")

Это мой celery.py, который находится в проекте cron:

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from celery.schedules import crontab


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


app = Celery('cron')


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


app.autodiscover_tasks()

app.conf.beat_schedule = {
    'check-articles-every-day': {
        'task': 'account.tasks.check_articles_and_send_slack_message',
        'schedule': crontab(hour=10, minute=45),  # Run at 10:45 a.m. every day
    },
}

Это код celery в файле settings.py, который находится в проекте cron:

CELERY_BROKER_URL = 'redis://127.0.0.1:6379'

# settings.py

CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379'

а это init.py, который находится в проекте под названием cron:

from .celery import app as celery_app

__all__ = ("celery_app",)

Вот команды celery, которые я использую :

celery -A cron worker --pool=solo -l info

celery -A cron beat -l info

celery -A cron call account.tasks.check_articles_and_send_slack_message

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