我目前正在研究一个烧瓶应用程序(Python 3.6),并希望集成芹菜,因为我有多个长时间运行的后台任务。
编辑:芹菜4.1
集成没有问题,芹菜任务正确执行,但我无法访问正在运行的任务的当前状态。
芹菜,烧瓶设置:
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)
芹菜任务:
@celery_app.task(bind=True, name="server.tasks.update_trading_pair")
def update_trading_pair(self, exchange, currency_a, currency_b):
print(exchange, currency_a, currency_b)
time.sleep(50)
调用任务并将值存储在字典中:
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
这是错误提出。当我只打印结果对象时,它只打印task_id。如果我尝试检索当前状态,我会引发以下异常:
TypeError: sequence item 1: expected a bytes-like object, AsyncResult found
解释:
当你打电话给你的任务:
task_id = update_trading_pair.delay(exchange, currency_a, currency_b)
你的变量task_id
是AsyncResult
的一个实例,它不是一个字符串。
所以,你的变量TASK_STATES[market.__id__()]
也是AsyncResult
的一个实例,而它应该是一个字符串。
然后你试图用它实例化一个AsyncResult
对象
result = update_trading_pair.AsyncResult(TASK_STATES[market.__id__()])
所以你用另一个AsyncResult
对象实例化一个AsyncResult
对象,而它应该用一个字符串实例化。
也许你的混乱来自你的print(task_id)
,它会显示一个字符串,但当你这样做时,在引擎盖下调用__str__
对象的AsyncResult
方法,如果你在源代码here中查看它,
def __str__(self):
"""`str(self) -> self.id`."""
return str(self.id)
它只是打印你的id
对象的task_id
属性。
解决方案
您可以通过执行TASK_STATES[id] = task_id.id
或通过执行来修复它
result = update_trading_pair.AsyncResult(str(TASK_STATES[market.__id__()]))