我不喜欢current_app的建议。
你的celery对象需要访问应用程序上下文。我在网上找到了一些有关使用工厂功能创建Celery对象的信息。以下示例在没有消息代理的情况下进行了测试。
#factory.pyfrom celery import Celeryfrom config import configdef 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.pyfrom factory import create_celery_appfrom celery.signals import task_prerunfrom flask import gcelery = create_celery_app()@task_prerun.connectdef 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
