监控数据变化

在本页

概述

示例数据
打开更改流
修改更改流输出
修改watch()行为
包含预图像和后图像
附加信息
API文档

概述

在本指南中，您可以了解如何使用更改流来监控数据库的实时更改。更改流是MongoDB服务器的一项功能，允许您的应用程序订阅集合、数据库或部署上的数据更改。

示例数据

本指南中的示例使用来自sample_restaurants.restaurants 集合的Atlas 示例数据集。要了解如何创建免费的 MongoDB Atlas 集群和加载数据集，请参阅PyMongo 入门.

打开变更流

要打开变更流，调用 watch() 方法。你调用 watch() 方法的实例决定了变更流监听的事件范围。你可以在以下类上调用 watch() 方法

MongoClient：用于监控 MongoDB 部署的所有变更
Database：用于监控数据库中所有集合的变更
Collection：用于监控集合的变更

以下示例在 restaurants 集合上打开变更流，并将更改实时输出

database = client["sample_restaurants"]
collection = database["restaurants"]
with collection.watch() as stream:
    for change in stream:
        print(change)

要开始监听变更，运行应用程序。然后在另一个应用程序或 shell 中修改 restaurants 集合。以下示例更新了一个具有 name 字段值为 Blarney Castle 的文档

database = client["sample_restaurants"]
collection = database["restaurants"]
query_filter = { "name": "Blarney Castle" }
update_operation = { '$set' : 
    { "cuisine": "Irish" }
}
result = collection.update_one(query_filter, update_operation)

当你更新集合时，变更流应用程序会实时打印更改。打印的更改事件类似于以下内容

{'_id': {'_data': '...'}, 'operationType': 'update', 'clusterTime': Timestamp(...), 'wallTime': datetime.datetime(...),
'ns': {'db': 'sample_restaurants', 'coll': 'restaurants'}, 'documentKey': {'_id': ObjectId('...')},
'updateDescription': {'updatedFields': {'cuisine': 'Irish'}, 'removedFields': [], 'truncatedArrays': []}}

修改更改流输出

您可以将 pipeline 参数传递给 watch() 方法来修改更改流输出。此参数允许您仅监视指定的更改事件。将参数格式化为表示聚合阶段的对象列表。

您可以在 pipeline 参数中指定以下阶段

$addFields
$match
$project
$replaceRoot
$replaceWith
$redact
$set
$unset

以下示例使用 pipeline 参数打开仅记录更新操作的更改流

change_pipeline = { "$match": { "operationType": "update" }},
with collection.watch(pipeline=change_pipeline) as stream:
    for change in stream:
        print(change)

要了解更多关于修改更改流输出的信息，请参阅 MongoDB 服务器手册中的修改更改流输出部分。

修改 `watch()` 行为

watch() 方法接受可选参数，这些参数表示您可以使用它们来配置操作的选项。如果您未指定任何选项，驱动程序不会自定义操作。

以下表格描述了您可以设置以自定义 watch() 行为的选项

属性	描述
`pipeline`	一个表示修改更改流输出的聚合管道阶段的列表。
`full_document`	指定在更改后是否显示整个文档，而不是仅显示对文档所做的更改。要了解更多关于此选项的信息，请参阅包含预映像和后映像。
`full_document_before_change`	指定是否显示更改之前的完整文档，而不是仅显示对文档所做的更改。要了解更多关于此选项的信息，请参阅包含预映像和后映像。
`resume_after`	指示`watch()`在恢复令牌指定的操作后继续返回更改。每个更改流事件文档都包含一个作为`_id`字段的恢复令牌。传递代表您想要恢复的操作的更改事件文档的整个`_id`字段。 `resume_after`与`start_after`和`start_at_operation_time`互斥。
`start_after`	指示`watch()`在恢复令牌指定的操作后开始新的更改流。允许在无效事件后恢复通知。每个更改流事件文档都包含一个作为`_id`字段的恢复令牌。传递代表您想要恢复的操作的更改事件文档的整个`_id`字段。 `start_after`与`resume_after`和`start_at_operation_time`互斥。
`start_at_operation_time`	指示`watch()`仅返回在指定时间戳之后发生的事件。 `start_at_operation_time`与`resume_after`和`start_after`互斥。
`max_await_time_ms`	服务器等待新数据更改报告给更改流游标的时间，以毫秒为单位，然后返回空批次。默认为1000毫秒。
`show_expanded_events`	从MongoDB服务器v6.0开始，更改流支持数据定义语言（DDL）事件（如`createIndexes`和`dropIndexes`事件）的更改通知。要在更改流中包含扩展事件，创建更改流游标并将此参数设置为`True`。
`batch_size`	每个批次从MongoDB集群的响应中返回的最大更改事件数。
`collation`	用于更改流游标的排序规则。
`session`	`ClientSession`的实例。
`comment`	要附加到操作的注释。

包含预映像和后映像

重要

您只能在部署使用 MongoDB v6.0 或更高版本的情况下，在集合上启用预图像和后图像。

默认情况下，当您对一个集合执行操作时，对应的变化事件仅包含该操作修改的字段的增量。要查看变化前后的完整文档，请在 watch() 方法中指定 full_document_before_change 或 full_document 参数。

预图像 是变化前文档的完整版本。要包括预图像在变化流事件中，将 full_document_before_change 参数设置为以下值之一

whenAvailable：仅当预图像可用时，变化事件才包含修改文档的预图像。
required：变化事件包含修改文档的预图像。如果预图像不可用，驱动程序将引发错误。

后图像 是变化后文档的完整版本。要包括后图像在变化流事件中，将 full_document 参数设置为以下值之一

updateLookup：变化事件包含从变化后某个时间点开始变化的整个文档的副本。
whenAvailable：仅当后图像可用时，变化事件才包含修改文档的后图像。
required：变化事件包含修改文档的后图像。如果后图像不可用，驱动程序将引发错误。

以下示例在集合上调用 watch() 方法，并通过指定 fullDocument 参数包括已更新文档的后图像

database = client["sample_restaurants"]
collection = database["restaurants"]
with collection.watch(full_document='updateLookup') as stream:
    for change in stream:
        print(change)

随着变化流应用程序的运行，使用前面的更新示例通过更新 restaurants 集合中的文档打印出类似于以下的变化事件

{'_id': {'_data': '...'}, 'operationType': 'update', 'clusterTime': Timestamp(...), 'wallTime': datetime.datetime(...),
 'fullDocument': {'_id': ObjectId('...'), 'address': {...}, 'borough': 'Queens',
 'cuisine': 'Irish', 'grades': [...], 'name': 'Blarney Castle', 'restaurant_id': '40366356'},
 'ns': {'db': 'sample_restaurants', 'coll': 'restaurants'}, 'documentKey': {'_id': ObjectId('...')},
 'updateDescription': {'updatedFields': {'cuisine': 'Irish'}, 'removedFields': [], 'truncatedArrays': []}}

有关预图像和后图像的更多信息，请参阅 MongoDB 服务器手册中的带有文档预图像和后图像的变化流。

概述

示例数据

打开变更流

修改更改流输出

修改 watch() 行为

包含预映像和后映像

重要

更多信息

API 文档

修改 `watch()` 行为