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

Question

Я пытаюсь перевернуть булевский флаг для определенных типов объектов в моей базе данных, используя sqlalchemy + celery beats. Но как мне получить доступ к моему orm из файла 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() 

Я использую команду сельдерей биений для запуска этого:

сельдерея -A задачи работника -l INFO --beat

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

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

Какие указывает на линию current_app.app_context()

Если я удалю текущий _app.app_context линия() Я получаю следующее сообщение об ошибке:

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

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

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

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

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

Answer 1

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

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

Или вы можете использовать шаблон, например, как описаны в the Flask docs подкласс 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)