Change Streams によるデータの監視

Overview

このガイドでは、変更ストリームを使用してデータベースへのリアルタイムの変更を監視する方法を学習できます。 変更ストリームは、アプリケーションがコレクション、データベース、または配置のデータ変更をサブスクライブできる MongoDB Server の機能です。

変更ストリームを開く

変更ストリームを開くには、 watch()メソッドを呼び出します。 watch()メソッドを呼び出す インスタンスによって、変更ストリームがリッスンするイベントの範囲が決まります。 次のクラスでwatch()メソッドを呼び出すことができます。

• MongoClient: MongoDB 配置のすべての変更を監視

• Database: データベース内のすべてのコレクションの変更を監視するには

• Collection: コレクションの変更をモニターするには

次の例では、restaurantsコレクションの変更ストリームを開き、変更が発生に応じて出力します。対応するコードを表示するには、Synchronous タブまたは Asynchronousタブを選択します。

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

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

変更の監視を開始するには、アプリケーションを実行します。次に、別のアプリケーションまたはシェルで、restaurantsコレクションを変更します。 次の例では、nameフィールドの値が Blarney Castle であるドキュメントを更新します。対応するコードを表示するには、Synchronous タブまたは Asynchronousタブを選択します。

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)

database = client["sample_restaurants"]
collection = database["restaurants"]
query_filter = { "name": "Blarney Castle" }
update_operation = { '$set' : 
    { "cuisine": "Irish" }
}
result = await 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 パラメータを使用して、アップデート操作のみを記録する変更ストリームを開きます。対応するコードを表示するには、Synchronous タブまたは Asynchronousタブを選択します。

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

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

変更ストリーム出力の変更の詳細については、MongoDB Server マニュアルの「 変更ストリーム出力 の変更 」セクションを参照してください。

watch() 動作を変更する

watch()メソッドは、操作を構成するために使用できるオプションを表す任意のパラメーターを受け入れます。 オプションを指定しない場合、ドライバーは操作をカスタマイズしません。

次の表では、 watch()の動作をカスタマイズするために設定できるオプションについて説明します。

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

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

{'_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': []}}

詳細情報

Change Streams変更ストリームの詳細については、MongoDB Server マニュアルの 「 ストリーム」 を参照してください。