Как получить доступ к форме с задачами сельдерея?

Я пытаюсь перевернуть логический флаг для определенных типов объектов в моей базе данных, используя sqlalchemy + celery beats. Но как мне получить доступ к моей форме из файла tasks.py?

from models import Book
from celery.decorators import periodic_task
from application import create_celery_app

celery = create_celery_app()
# Create celery: http://flask.pocoo.org/docs/0.10/patterns/celery/

# This task works fine
@celery.task
def celery_send_email(to,subject,template):
    with current_app.app_context():
        msg = Message(
            subject,
            recipients=[to],
            html=template,
            sender=current_app.config['MAIL_DEFAULT_SENDER']
        )
        return mail.send(msg)

#This fails
@periodic_task(name='release_flag',run_every=timedelta(seconds=10))
def release_flag():
    with current_app.app_context(): <<< #Fails on this line
        books = Book.query.all() <<<< #Fails here too
        for book in books:
          book.read = True
          book.save()

Я использую команду celery beat для запуска:

celery -A Tasks Worker -l INFO --beat

Но я получаю следующую ошибку:

raise RuntimeError('working outside of application context')
RuntimeError: working outside of application context

Что указывает на строку with current_app.app_context()

Если я удалю строку current_app.app_context(), я получу следующую ошибку:

RuntimeError: application not registered on db instance and no application bound to current context

Есть ли особый способ доступа к форме flask-sqlalchemy для задач сельдерея? Или был бы лучший подход к тому, что я пытаюсь сделать?

До сих пор единственным работающим обходным путем было добавление следующей строки после db.init_app(app) в моем шаблоне фабрики приложений:

db.app = приложение

Я следил за этим репозиторием, чтобы создать приложение для сельдерея https://github.com/mattupstate/overholt/blob/master/overholt/factory.py


flask flask-sqlalchemy celery celery-task celerybeat
person user805981    schedule 29.07.2015    source источник

Ответы (1)


arrow_upward
7
arrow_downward

Вы получаете эту ошибку, потому что current_app для работы требуется контекст приложения, но вы пытаетесь использовать его для настройки контекста приложения. Вам нужно использовать фактическое приложение для настройки контекста, тогда вы можете использовать current_app.

with app.app_context():
    # do stuff that requires the app context

Или вы можете использовать шаблон, подобный описанному в документах Flask, чтобы подкласс celery.Task, чтобы он знал о контексте приложения по умолчанию.

from celery import Celery

def make_celery(app):
     celery = Celery(app.import_name, 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

 celery = make_celery(app)
person davidism    schedule 01.08.2015