【发布时间】:2018-04-19 14:35:55
【问题描述】:
我正在尝试按照http://flask.pocoo.org/docs/0.12/patterns/celery/ 中的说明进行操作,以便可以在 celery 任务中执行烧瓶/socketIO 操作。但是,我的目录结构有点不同,而且我在导入方面没有任何运气。
我的目录结构如下:
├── app
│ ├── __init__.py
│ ├── __pycache__
│ ├── auth.py
│ ├── ctasks.py
│ ├── helper.py
│ ├── saml.py
│ ├── socks.py
│ ├── templates
│ ├── threads.py
│ └── views.py
├── app.py
├── config.py
├── requirements.txt
└── saml
├── dev
└── prod
我从app.py调用应用程序
from app import socketio, app
if __name__ == '__main__':
socketio.run(app, debug=True, port=443, ssl_context='adhoc')
__init__.py
from flask import Flask, request
from flask_socketio import SocketIO
from .ctasks import subtaskcaller, make_celery
from .helper import wait_to_finish
async_mode = None
app = Flask(__name__)
app.config.from_object('config')
socketio = SocketIO(app, async_mode=async_mode)
cel = make_celery(app)
from .auth import SamlManager
saml_manager = SamlManager()
saml_manager.init_app(app)
from app import views, socks, saml, helper, ctasks
ctasks.py
from celery import Celery
from config import *
from .helper import wait_to_finish, emitter
import time
from app import cel
def make_celery(app):
c = Celery(app.import_name, backend=CELERY_RESULT_BACKEND, broker=CELERY_BROKER_URL)
c.conf.update(app.config)
taskbase = c.Task
class ContextTask(taskbase):
abstract = True
def __call__(self, *args, **kwargs):
with app.app_context():
return taskbase.__call__(self, *args, **kwargs)
c.Task = ContextTask
return c
@cel.task(name='tasks.tester', serializer='pickle')
def tester():
emitter('emit from subsubtask')
for i in range(1, 50):
time.sleep(1)
print('test {0}'.format(i))
x = True
return x
@cel.task(name='task.subtaskcaller', serializer='pickle')
def subtaskcaller():
emitter('emit from subtask')
finished = tester.delay()
wait_to_finish(finished)
return finished
我在尝试从ctasks.py 中的应用程序导入 cel 时遇到错误:
ImportError: cannot import name 'cel'
【问题讨论】:
-
我相信我正确使用了单位并且过去使用过与此类似的导入。我猜我有问题,循环导入或导入依赖问题。我的原始代码中有错字并已更新。