我目前正在开发一个flask应用程序(python3.6),并希望集成celeri,因为我有多个长期运行的后台任务。在
编辑:芹菜4.1
集成没有问题,celery任务也正确执行,但我无法访问正在运行的任务的当前状态。在
芹菜、烧瓶设置:
def make_celery(app):
celery = Celery(app.import_name,
backend=app.config["result_backend"],
broker=app.config["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 = Flask(__name__)
app.config["broker_url"] = "redis://localhost:6379"
app.config["result_backend"] = "redis://localhost:6379"
app.config["DB"] = "../pyhodl.sqlite"
celery_app = make_celery(app)
芹菜任务:
^{pr2}$调用任务并将值存储在字典中:
task_id = update_trading_pair.delay(exchange, currency_a, currency_b)
print("NEW TASK")
print(task_id)
id = exchange_mnemonic + "_" + currency_a + "_" + currency_b
TASK_STATES[id] = task_id
获取任务状态:
result = update_trading_pair.AsyncResult(TASK_STATES[market.__id__()])
print(result.state)
print(result) # works but only prints the task_id
这是因为出现了错误。当我只打印result对象时,它只打印任务的标识。如果我试图检索当前状态,则会引发以下异常:
TypeError: sequence item 1: expected a bytes-like object, AsyncResult found
说明:
当您调用任务时:
您的变量
task_id
是AsyncResult
的实例,它不是字符串。在因此,您的变量
TASK_STATES[market.__id__()]
也是AsyncResult
的实例,而它应该是一个字符串。在然后尝试用它实例化一个
^{pr2}$AsyncResult
对象因此,您使用另一个
AsyncResult
对象实例化一个AsyncResult
对象,而它应该用字符串实例化。在也许您的困惑来自于您的
print(task_id)
,它向您显示了一个字符串,但是当您这样做时,AsyncResult
对象的__str__
方法被调用,如果您在源代码here中查看它它只是打印
id
对象的id
属性。在解决方案:
您可以通过执行
TASK_STATES[id] = task_id.id
或相关问题 更多 >
编程相关推荐