Django и Celery в разных репозиториях

У меня есть два проекта которые будут в разных репозиториях, 1-й это django, 2-ой celery Как мне отправлять задачу из django в celery?

Код django:

__init__.py:

from .celery import app as celery_app

__all__ = ('celery_app',)

settings.py:

CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0',
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0',
CELERY_TIMEZONE = TIME_ZONE
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
DJANGO_CELERY_BEAT_TZ_AWARE = False

celery.py:

import os
from celery import Celery

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

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

tasks.py:

from celery import shared_task


@shared_task(name="test_task")
def test_task():
    return "Hello from Django!"

Код celery:

celery_tasks/celery.py:

from celery import Celery

app = Celery('celery_tasks')

app.conf.update(
    broker_url='redis://127.0.0.1:6379/0',
    result_backend='redis://127.0.0.1:6379/0',
    task_serializer='json',
    accept_content=['json'],
    timezone='Europe/Moscow',
    broker_connection_retry_on_startup=True
)


app.autodiscover_tasks(['celery_tasks'])

celery_tasks/tasks.py:

from celery import Celery

app = Celery('celery_tasks')


@app.task(name="test_task")
def test_task():
    return "Hello from Celery!"

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