观看 MongoDB 在 Python 中异步更改流

我如何设置一个 python 服务来（异步）监视 mongodb 的更改流。

我能找到的一切mongodb.com https://www.mongodb.com/developer/quickstart/python-change-streams/ and pymongo 文档 https://pymongo.readthedocs.io/en/stable/api/pymongo/change_stream.html以下两种方法似乎还没有真正准备好投入生产：

访问 mongodb.com：

import os
import pymongo
from bson.json_util import dumps
client = pymongo.MongoClient('<YOUR-MONGO-CONNECT-STRING>')
change_stream = client.changestream.collection.watch()
for change in change_stream:
    print(dumps(change))

方法 pymongo 文档：

with db.collection.watch() as stream:
    while stream.alive:
        change = stream.try_next()
        print("Current resume token: %r" % (stream.resume_token,))
        if change is not None:
            print("Change document: %r" % (change,))
            continue
        time.sleep(10)

我考虑了一个使用 watch 函数作为事件循环回调的解决方案。 有人知道这个的实现吗？

获取所有 MongoDB 更新通知的简单方法如下：

import pymongo

client = pymongo.MongoClient("mongodb+srv://YOUR-MONGO-CONNECT-STRING" % (
            mongo_user, mongo_pass, mongo_db_name))

 option={ 'full_document':'updateLookup' }

 change_stream = client.mongo_db_name.mongo_db_collection.watch([{"$match" : { "operationType" : "update" }}], **option)
 for change in change_stream:
     print(dumps(change))
     print('')

您需要更换

• YOUR-MONGO-CONNECT-STRING：包含您的 MongoDB 连接 URL
• mongo_user：使用您的用户名
• mongo_pass：使用您的密码
• mongo_db_name：您想要观看的数据库名称
• mongo_db_collection：带有您想要观看的集合名称

此外，您可以添加简历令牌以处理您将要处理的情况，例如出现网络错误。使用恢复令牌，流将再次连接，并且您将从丢失连接的那一刻收到通知。

import pymongo
from pymongo import errors

client = pymongo.MongoClient("mongodb+srv://YOUR-MONGO-CONNECT-STRING" % (
            mongo_user, mongo_pass, mongo_db_name))

try:
    resume_token = None
    pipeline = [{'$match': {'operationType': 'update'}}]
    with client.mongo_db_name.mongo_db_collection.watch(pipeline) as stream:
        for update_change in stream:
            print(update_change)
            resume_token = stream.resume_token
except pymongo.errors.PyMongoError:
    if resume_token is None:
       logging.error('...')
    else:
        with client.mongo_db_name.mongo_db_collection.watch(pipeline, resume_after=resume_token) as stream:
            for update_change in stream:
                print(update_change)