python - Django、带递归的 Celery 和 Twitter API

Question

我正在使用 Django 1.4 和 Celery 3.0 (rabbitmq) 构建一个任务集合，用于获取和缓存对 Twitter API 1.1 的查询。我试图实现的一件事是任务链，最后一个任务根据到目前为止的响应和最近检索到的响应中的响应数据，递归调用两个节点后的任务。具体来说，这允许应用程序遍历用户时间线(最多 3200 条推文)，同时考虑到任何给定请求最多只能产生 200 条推文(Twitter API 的限制)。

我的tasks.py关键组件可以看here ，但在粘贴之前，我将展示我从我的 Python shell 调用的链(但最终将通过用户在最终 Web 应用程序中的输入启动)。鉴于:

>>request(twitter_user_id='#1010101010101#, 
  total_requested=1000, 
  max_id = random.getrandbits(128) #e.g. arbitrarily large number)

我打电话:

>> res = (twitter_getter.s(request) | 
        pre_get_tweets_for_user_id.s() | 
        get_tweets_for_user_id.s() | 
        timeline_recursor.s()).apply_async()

关键是 timeline_recursor 可以启动数量可变的 get_tweets_for_user_id 子任务。当 timeline_recursor 处于其基本情况时，它应该返回此处定义的响应字典:

@task(rate_limit=None)
def timeline_recursor(request):
    previous_tweets=request.get('previous_tweets', None) #If it's the first time through, this will be None
    if not previous_tweets:
        previous_tweets = [] #so we initiate to empty array
    tweets = request.get('tweets', None) 

    twitter_user_id=request['twitter_user_id']
    previous_max_id=request['previous_max_id']
    total_requested=request['total_requested']
    pulled_in=request['pulled_in']

    remaining_requested = total_requested - pulled_in
    if previous_max_id:
        remaining_requested += 1 #this is because cursored results will always have one overlapping id

    else:
        previous_max_id = random.getrandbits(128) # for first time through loop

    new_max_id = min([tweet['id'] for tweet in tweets])
    test = lambda x, y: x<y

    if remaining_requested < 0:  #because we overshoot by requesting batches of 200
        remaining_requested = 0

    if tweets:
        previous_tweets.extend(tweets)

    if tweets and remaining_requested and (pulled_in > 1) and test(new_max_id, previous_max_id):

        request = dict(user_pk=user_pk,
                    twitter_user_id=twitter_user_id,
                    max_id = new_max_id,
                    total_requested = remaining_requested,
                    tweets=previous_tweets)

        #problem happens in this part of the logic???

        response = (twitter_getter_config.s(request) | get_tweets_for_user_id.s() | timeline_recursor.s()).apply_async()

    else: #if in base case, combine all tweets pulled in thus far and send back as "tweets" -- to be 
          #saved in db or otherwise consumed
        response = dict(
                    twitter_user_id=twitter_user_id,
                    total_requested = total_requested,
                    tweets=previous_tweets)
    return response

因此，我对 res.result 的预期响应是一个字典，其中包含 Twitter 用户 ID、请求的推文数量以及在连续调用中拉入的推文集。然而，在递归任务领域并不是一切都很好。当我运行上面确定的链时，如果我在启动链后立即输入 res.status，它表示“成功”，即使在我的 celery worker 的日志 View 中，我可以看到链式递归调用到 twitter api 正在按预期进行，使用正确的参数。即使正在执行链式任务，我也可以立即运行 result.result。 res.result 产生一个 AsyncResponse 实例 ID。即使在递归链接的任务完成运行后，res.result 仍然是一个 AsyncResult id。

另一方面，我可以通过转至 res.result.result.result.result['tweets'] 访问我的完整推文集。我可以推断每个链式子任务确实在发生，我只是不明白为什么 res.result 没有预期的结果。当 timeline_recursor 获得其基本情况时应该发生的递归返回似乎没有按预期传播。

有什么可以做的想法吗？ Celery 中的递归可以变得非常强大，但至少对我来说，我们应该如何考虑使用 Celery 的递归和递归函数以及这如何影响链式任务中返回语句的逻辑并不完全清楚。

很乐意根据需要进行澄清，并提前感谢您的任何建议。

Answer 1

apply_async 返回什么(在对象类型中)？

我不知道 celery，但在 Twisted 和许多其他异步框架中......调用类似的东西会立即返回(通常是 True 或者可能是一个可以跟踪状态的对象)作为任务被推迟到队列中。

再次，不知道 celery ，我猜这是发生了:

您是:将 response 立即定义为异步延迟的 task，然后尝试对其采取行动，就好像结果已经出现一样

您想成为:定义一个 callback 例程以在任务完成后运行结果并返回一个值

查看 celery 文档，apply_async 通过 link 接受回调 - 我找不到任何试图从中捕获返回值的示例。