Я пытаюсь следовать этому Фоновые задачи на основе сельдерея , чтобы создать настройки сельдерея для простого приложения.
В моем task.py
from celery import Celery
def make_celery(app):
celery = Celery(app.import_name, backend=app.config['CELERY_RESULT_BACKEND'],
broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)
TaskBase = celery.Task
class ContextTask(TaskBase):
abstract = True
def __call__(self, *args, **kwargs):
with app.app_context():
return TaskBase.__call__(self, *args, **kwargs)
celery.Task = ContextTask
return celery
Этот метод работает в app.py
приложения основной колбы.
from flask import Flask
flask_app = Flask(__name__)
flask_app.config.update(
CELERY_BROKER_URL='redis://localhost:6379',
CELERY_RESULT_BACKEND='redis://localhost:6379'
)
celery = make_celery(flask_app)
@celery.task()
def add_together(a, b):
return a + b
Мой пример использования: я хочу создать еще один модуль helpers.py
, в котором я могу определить наборы асинхронныхклассы.Чтобы отделить методы на основе сельдерея и сделать их модульными.
То, что я сделал, это вызвал модуль task.py
для другого модуля helpers.py
, чтобы создать класс AsyncMail
для обработки фоновой работы действий электронной почты..
from task import make_celery
class AsyncMail(object):
def __init__(self, app):
"""
:param app: An instance of a flask application.
"""
self.celery = make_celery(app)
def send(self, msg):
print(msg)
Теперь, как я могу получить доступ к атрибуту self.celery
, чтобы быть декоратором для любого метода класса?
@celery.task()
def send(self, msg):
print(msg)
Если это невозможно, какие другие альтернативные шаги для того, чтобыдостиг этой проблемы?