Tornado 协程逐渐从另一个协程中产生

Question

我想重构我的 Tornado 应用程序的一部分，所以我创建了一个特殊的函数来返回电话号码：

@gen.coroutine
def get_phones(self, listname):
    phones = []
    logging.info("fetching phones")
    cursor = self._mongo.contacts.aggregate(self.get_query(
        subscription_filter={
            "$ne": [{"$ifNull": ["$$subscription.events.{listname}", None]}, None]
        },
        handler_filter={
            "handler.meta.is_active": True,
            "handler.meta.type": "phone"
        }
    ))
    try:
        while (yield cursor.fetch_next):
            contact = cursor.next_object()
            logging.info(contact)
            try:
                phones += [handler['subject'] for handler in contact['handlers']]
                if len(phones) > 50:
                    yield phones
                    phones = []
            except Exception:
                self._logger.warning("Could not get phone no")
    except Exception:
        phones = []
        logging.warning("Could not fetch contacts")

    if len(phones) > 0:
        yield phones

我想要实现的是从我的数据库中异步获取最多 50 个联系人的批次并将它们返回给调用协程。

这是我的调用协程：

@gen.coroutine
def on_heartbeat_status_update(self, status):
    phonegen = self.get_phones("ews:admin")
    logging.info(phonegen)
    while True:
        phones = yield phonegen
        logging.info(phones)
        if phones is None:
            break
        logging.info(len(phones))

它不工作。 “电话”总是无。有人可以建议实现这一目标的正确方法吗？谢谢！

Answer 1

您必须使用 Python 3.6 原生协程才能使其正常工作。这是第一个在同一函数中同时支持yield 和await 的Python 版本。没有这个，系统就无法区分使用yield 作为协程和生成结果作为生成器。

将@gen.coroutine def 替换为async def 并在get_phones 中使用await cursor.fetch_next 和yield phones。然后你可以在on_heartbeat_status_update 中使用async for phones in self.get_phones(...)。