进口烧瓶中的芹菜蓝图

Question

我有一个具有MVC结构的Flask应用程序：

my_app
├── server.py
├── requirements.txt
├── models
│   ├── __init__.py
└── model.py
├── controllers
├── __init__.py
├── client_controllers
└──controller.py
└── another_controller.py
└── templates

我使用蓝图将服务器代码拆分为"控制器"，所以我有这样的东西：

server.py：

from flask import Flask
from celery import Celery
from controllers.client_controllers.controller import controller
app = Flask(__name__)
app.secret_key = 'SECRET'
app.register_blueprint(controller)
# Celery Configuration
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.config.update(
CELERY_BROKER_URL='redis://localhost:6379',
CELERY_RESULT_BACKEND='redis://localhost:6379'
)
celery = make_celery(app)

if __name__ == "__main__":
app.run(host='0.0.0.0', debug=True)

控制器.py：

from flask import Blueprint, render_template, json, request, redirect, url_for, abort, session

controller = Blueprint('controller', __name__,
template_folder='templates/')
@celery.task()
def add_together(a, b):
return a + b
@controller.route('/add', methods=['GET'])
def add():
result = add_together.delay(23, 42)
result.wait()
return 'Processing'

正如您可能注意到的，芹菜没有导入到控制器中，因为我不知道如何将芹菜实例从服务器.py导入到我的控制器.py中而不会出现错误，我一直在尝试：

from ...server import celery
from ..server import celery
...etc

但仍因错误而失败。

Answer 1

发生烧瓶错误RuntimeError: Working outside of application context.是因为您不在烧瓶application_context()中。

您应该使用celeceshared_task，这是您在MVC结构下所需要的。

celery_flask/
├── celery_tasks
│   ├── app_tasks.py
│   ├── __init__.py
├── celery_worker.py
├── controllers
│   ├── __init__.py
│   ├── some_controller.py
├── __init__.py
└── server.py

脚本app_tasks.py

#=====================
#   app_tasks.py
#=====================
from __future__ import absolute_import, unicode_literals
from celery import shared_task
@shared_task(name='celery_tasks.add_together')
def add_together(x, y):
return x + y

@shared_task装饰器返回一个始终指向活动Celery实例的代理：

>>> from celery import Celery, shared_task
>>> @shared_task
... def add_together(x, y):
...     return x + y
...
>>> app1 = Celery(broker='amqp://')
>>> add_together.app is app1
True
>>> app2 = Celery(broker='redis://')
>>> add_together.app is app2
True

定义任务后，您可以使用对Celery应用程序的引用来调用它们。这个芹菜应用程序可能是flask application_context()的一部分。示例：

脚本server.py

from __future__ import absolute_import
from flask import Flask
from celery import Celery
from controllers.some_controller import controller
flask_app = Flask(__name__)
flask_app.secret_key = 'SECRET'
flask_app.register_blueprint(controller)
# Celery Configuration
def make_celery( app ):
celery = Celery('flask-celery-app', backend=app.config['CELERY_RESULT_BACKEND'],
broker=app.config['CELERY_BROKER_URL'],
include=['celery_tasks.app_tasks'])
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
def list_celery_task( ):
from celery.task.control import inspect
i = inspect()
i.registered_tasks()
from itertools import chain
t = set(chain.from_iterable( i.registered_tasks().values() ))
print "registered_tasks={}".format( t )
#======================================
#                MAIN
#======================================
flask_app.config.update(
CELERY_BROKER_URL='redis://localhost:6379',
CELERY_RESULT_BACKEND='redis://localhost:6379'
)
celery = make_celery(flask_app)
flask_app.celery = celery
list_celery_task( )

if __name__ == "__main__":
flask_app.run(host='0.0.0.0', debug=True)

脚本some_controller.py

#============================
#   some_controller.py
#============================
from __future__ import absolute_import
from flask import Blueprint
from flask import current_app
controller = Blueprint('controller', __name__,
template_folder='templates/')
@controller.route('/add', methods=['GET'])
def add():
print "calling add"
result = current_app.celery.send_task('celery_tasks.add_together',args=[12,6])
r = result.get()
print 'Processing is {}'.format( r )
return 'Processing is {}'.format( r )

最后，启动工人消耗任务：

celery -A celery_worker worker --loglevel=DEBUG

脚本cele_worker.py

#============================
#   celery_worker.py
#============================
from __future__ import absolute_import
from celery import Celery
# Celery Configuration
def make_celery():
celery = Celery('flask-celery-app', backend='redis://localhost:6379',
broker='redis://localhost:6379',
include=['celery_tasks.app_tasks'])
return celery
celery = make_celery()
print "tasks={}".format( celery.tasks.keys() )

Answer 2

一个选项是将芹菜实例分配给应用程序实例，然后通过flask的current_app访问它。

在server.py中，只需添加：

celery = make_celery(app)
app.celery = celery

然后你可以在你的控制器中访问这个。py:

from flask import current_app
@current_app.celery.task()
def add_together(a, b):
return a + b