Колба с помощью create_app, SQLAlchemy и Celery

Ответ 1

Я был у вас с советом current_app.

Ваш объект сельдерея нуждается в доступе к контексту приложения. Я нашел некоторую информацию в Интернете о создании объекта Celery с помощью функции factory. Пример ниже проверяется без брокера сообщений.

#factory.py
from celery import Celery
from config import config

def create_celery_app(app=None):
    app = app or create_app(config)
    celery = Celery(__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

и в tasks.py:

#tasks.py
from factory import create_celery_app
from celery.signals import task_prerun
from flask import g

celery = create_celery_app()

@task_prerun.connect
def celery_prerun(*args, **kwargs):
    #print g
    with celery.app.app_context():
    #   # use g.db
       print g

@celery.task()
def do_some_stuff():
    with celery.app.app_context():
        # use g.db
        g.user = "test"
        print g.user

Некоторые ссылки:

Флажок для создания экземпляра Celery с функцией factory

Приложение, использующее как приложение factory, так и сельдерей

Источник для указанного приложения factory.py

Источник для приложений tasks.py

Ответ 2

Вот решение, которое работает с шаблоном фабрики приложений фляги, а также создает задачу сельдерея с контекстом, без необходимости явно использовать app.app_context() в задачах. В моем приложении действительно сложно получить этот объект приложения, избегая циклического импорта, но это решает его. Это также хорошо для последней версии сельдерея 4.2 на момент написания.

Состав:

repo_name/
    manage.py
    base/
    base/__init__.py
    base/app.py
    base/runcelery.py
    base/celeryconfig.py
    base/utility/celery_util.py
    base/tasks/workers.py

Таким образом, base - это основной пакет приложения в этом примере. В base/__init__.py мы создаем экземпляр сельдерея, как base/__init__.py ниже:

from celery import Celery
celery = Celery('base', config_source='base.celeryconfig')

Файл base/app.py содержит фабрику приложений фляги create_app и обратите внимание на init_celery(app, celery) он содержит:

from base import celery
from base.utility.celery_util import init_celery

def create_app(config_obj):
    """An application factory, as explained here:
    http://flask.pocoo.org/docs/patterns/appfactories/.
    :param config_object: The configuration object to use.
    """
    app = Flask('base')
    app.config.from_object(config_obj)
    init_celery(app, celery=celery)
    register_extensions(app)
    register_blueprints(app)
    register_errorhandlers(app)
    register_app_context_processors(app)
    return app

Переход к содержанию base/runcelery.py:

from flask.helpers import get_debug_flag
from base.settings import DevConfig, ProdConfig
from base import celery
from base.app import create_app
from base.utility.celery_util import init_celery
CONFIG = DevConfig if get_debug_flag() else ProdConfig
app = create_app(CONFIG)
init_celery(app, celery)

Далее файл base/celeryconfig.py (как пример):

# -*- coding: utf-8 -*-
"""
Configure Celery. See the configuration guide at ->
http://docs.celeryproject.org/en/master/userguide/configuration.html#configuration
"""

## Broker settings.
broker_url = 'pyamqp://guest:[email protected]:5672//'
broker_heartbeat=0

# List of modules to import when the Celery worker starts.
imports = ('base.tasks.workers',)

## Using the database to store task state and results.
result_backend = 'rpc'
#result_persistent = False

accept_content = ['json', 'application/text']

result_serializer = 'json'
timezone = "UTC"

# define periodic tasks / cron here
# beat_schedule = {
#    'add-every-10-seconds': {
#        'task': 'workers.add_together',
#        'schedule': 10.0,
#        'args': (16, 16)
#    },
# }

Теперь определите init_celery в файле base/utility/celery_util.py:

# -*- coding: utf-8 -*-

def init_celery(app, celery):
    """Add flask app context to celery.Task"""
    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

Для работников в base/tasks/workers.py:

from base import celery as celery_app
from flask_security.utils import config_value, send_mail
from base.bp.users.models.user_models import User

@celery_app.task
def send_welcome_email(email, user_id, confirmation_link):
    """Background task to send a welcome email with flask-security mail.
    You don't need to use with app.app_context() as Task has app context.
    """
    user = User.query.filter_by(id=user_id).first()
    print(f'sending user {user} a welcome email')
    send_mail(config_value('EMAIL_SUBJECT_REGISTER'),
              email,
              'welcome', user=user,
              confirmation_link=confirmation_link) 

@celery_app.task
def do_some_stuff():
    print(g)

Затем вам нужно запустить бит сельдерея и сельдерея в двух разных repo_name cmd из папки repo_name.

В одном celery -A base.runcelery:celery beat cmd сделайте celery -A base.runcelery:celery beat а другой celery -A base.runcelery:celery worker.

Затем выполните вашу задачу, для которой нужен контекст колбы. Должно сработать.