带create_app，SQLAlchemy和Celery的烧瓶 [英] Flask with create_app, SQLAlchemy and Celery
问题描述
我真的在努力为Flask，SQLAlchemy和Celery设置正确的设置。我进行了广泛的搜索并尝试了不同的方法，但似乎没有任何效果。我错过了应用程序上下文或无法运行工作程序，或者出现了其他一些问题。该结构非常通用，因此我可以构建一个更大的应用程序。
I'm really struggling to the get the proper setup for Flask, SQLAlchemy and Celery. I have searched extensively and tried different approaches, nothing really seems to work. Either I missed the application context or can't run the workers or there are some other problems. The structure is very general so that I can build a bigger application.
我正在使用：Flask 0.10.1，SQLAlchemy 1.0，Celery 3.1.13，我当前的设置如下：
I'm using: Flask 0.10.1, SQLAlchemy 1.0, Celery 3.1.13, my current setup is the following:
app / __ init __。py
app/__init__.py
#Empty
app / config.py
app/config.py
import os
basedir = os.path.abspath(os.path.dirname(__file__))
class Config:
@staticmethod
def init_app(app):
pass
class LocalConfig(Config):
DEBUG = True
SQLALCHEMY_DATABASE_URI = r"sqlite:///" + os.path.join(basedir,
"data-dev.sqlite")
CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'
config = {
"local": LocalConfig}
app / exstensions.py
app/exstensions.py
from flask.ext.sqlalchemy import SQLAlchemy
from celery import Celery
db = SQLAlchemy()
celery = Celery()
app / factory.py
app/factory.py
from extensions import db, celery
from flask import Flask
from flask import g
from config import config
def create_before_request(app):
def before_request():
g.db = db
return before_request
def create_app(config_name):
app = Flask(__name__)
app.config.from_object(config[config_name])
db.init_app(app)
celery.config_from_object(config)
# Register the blueprints
# Add the before request handler
app.before_request(create_before_request(app))
return app
app / manage.py
app/manage.py
from factory import create_app
app = create_app("local")
from flask import render_template
from flask import request
@app.route('/test', methods=['POST'])
def task_simple():
import tasks
tasks.do_some_stuff.delay()
return ""
if __name__ == "__main__":
app.run()
app / models.py
app/models.py
from extensions import db
class User(db.Model):
__tablename__ = "user"
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(128), unique=True, nullable=False)
app / tasks.py
app/tasks.py
from extensions import celery
from celery.signals import task_prerun
from flask import g, current_app
@task_prerun.connect
def close_session(*args, **kwargs):
with current_app.app_context():
# use g.db
print g
@celery.task()
def do_some_stuff():
with current_app.app_context():
# use g.db
print g
在文件夹应用程序中：
- 使用以下命令启动开发Web服务器：
python.exe manage.py - 以以下方式启动工作人员：
celery.exe worker -A任务
- starting the development webserver with:
python.exe manage.py - starting the workers with:
celery.exe worker -A tasks
我收到对我来说没有任何意义的导入错误。
我应该以不同的方式构造应用程序吗？最后，我想我想要一个非常基本的设置，例如将Flask与工厂模式一起使用，可以使用Flask-SQLAlchmey扩展，并需要一些需要访问数据库的工作人员。
I get an import error that doesn't make any sense to me. Should I structure the application differently? At the end I think I want a quite basic setup, e.g. Using Flask with the factory pattern, be able to use the Flask-SQLAlchmey extension and have some worker that needs to access the database.
我们非常感谢您的帮助。
Any help is highly appreciated.
启动芹菜工作者时会执行追溯。
Traceback is executed when starting the celery worker.
Traceback (most recent call last):
File "[PATH]\scripts\celery-script.py", line 9, in <module>
load_entry_point('celery==3.1.13', 'console_scripts', 'celery')()
File "[PATH]\lib\site-packages\celery\__main__.py", line 30, in main
main()
File "[PATH]\lib\site-packages\celery\bin\celery.py", line 81, in main
cmd.execute_from_commandline(argv)
File "[PATH]\lib\site-packages\celery\bin\celery.py", line 769, in execute_from_commandline
super(CeleryCommand, self).execute_from_commandline(argv)))
File "[PATH]\lib\site-packages\celery\bin\base.py", line 305, in execute_from_commandline
argv = self.setup_app_from_commandline(argv)
File "[PATH]\lib\site-packages\celery\bin\base.py", line 473, in setup_app_from_commandline
user_preload = tuple(self.app.user_options['preload'] or ())
AttributeError: 'Flask' object has no attribute 'user_options'
更新我会根据建议更改代码在评论中。该工作程序现在启动，但是在进行测试时，使用对 http://127.0.0.1:5000/test 的get请求。我得到以下回溯：
UPDATE I change the code according to the suggestion in the comment. The worker starts now up but when test it with a get request to http://127.0.0.1:5000/test. I get the following traceback:
Traceback (most recent call last):
File "[PATH]\lib\site-packages\celery\app\trace.py", line 230, in trace_task
args=args, kwargs=kwargs)
File "[PATH]\lib\site-packages\celery\utils\dispatch\signal.py", line 166, in send
response = receiver(signal=self, sender=sender, \**named)
File "[PATH]\app\stackoverflow\tasks.py", line 7, in close_session
with current_app.app_context():
File "[PATH]\lib\site-packages\werkzeug\local.py", line 338, in __getattr__
return getattr(self._get_current_object(), name)
File "[PATH]\lib\site-packages\werkzeug\local.py", line 297, in _get_current_object
return self.__local()
File "[PATH]\lib\site-packages\flask\globals.py", line 34, in _find_app
raise RuntimeError('working outside of application context')
RuntimeError: working outside of application context exc, exc_info.traceback)))
更新在Marteen的评论中，我更改了代码。当前工作版本位于： https://gist.github.com/anonymous/fa47834db2f4f3b8b257 。
欢迎任何进一步的改进或建议。
UPDATE Based on the comments from Marteen, I changed the code. The current working version is under: https://gist.github.com/anonymous/fa47834db2f4f3b8b257. Any further improvements or suggestions are welcome.
推荐答案
我不喜欢current_app的建议。
I was off with the current_app advice.
您的celery对象需要访问应用程序上下文。我在网上找到了一些有关使用工厂函数创建Celery对象的信息。以下示例在没有消息代理的情况下进行了测试。
Your celery object needs access to the application context. I found some information online about creating the Celery object with a factory function. Example below is tested without a message broker.
#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
以及在task.py中：
and in 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
某些链接：
这篇关于带create_app，SQLAlchemy和Celery的烧瓶的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
