Monitorear datos con flujos de cambios

Overview

En esta guía, aprenderá a usar un flujo de cambios para supervisar los cambios en tiempo real en su base de datos. Un flujo de cambios es una función de MongoDB Server que permite que su aplicación se suscriba a los cambios en los datos de una colección, base de datos o implementación.

Abre un flujo de cambios

Para abrir un flujo de cambios, llame al método watch(). La instancia en la que llama al método watch() determina el alcance de los eventos que el flujo de cambios escucha. Puede llamar al método watch() en las siguientes clases:

• MongoClient:Para monitorear todos los cambios en la implementación de MongoDB

• Database:Para monitorear los cambios en todas las colecciones de la base de datos

• Collection:Para monitorear cambios en la colección

El siguiente ejemplo abre un flujo de cambios en la colección restaurants y genera los cambios a medida que ocurren. Seleccione el Synchronous o pestaña Asynchronous para ver el código correspondiente:

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)

Para empezar a observar los cambios, ejecute la aplicación. Luego, en una aplicación o shell independiente, modifique la colección restaurants. El siguiente ejemplo actualiza un documento con un valor de campo name de Blarney Castle. Seleccione la pestaña Synchronous o Asynchronous para ver el código correspondiente:

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)

Al actualizar la colección, la aplicación de flujo de cambios imprime el cambio a medida que se produce. El evento de cambio impreso se parece al siguiente:

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

Modificar la salida del flujo de cambios

Puede pasar el parámetro pipeline al método watch() para modificar la salida del flujo de cambios. Este parámetro le permite observar únicamente los eventos de cambio especificados. Formatee el parámetro como una lista de objetos que representen cada uno una etapa de agregación.

Puede especificar las siguientes etapas en el parámetro pipeline:

• $addFields

• $match

• $project

• $replaceRoot

• $replaceWith

• $redact

• $set

• $unset

El siguiente ejemplo utiliza el parámetro pipeline para abrir un flujo de cambios que registra únicamente las operaciones de actualización. Seleccione la pestaña Synchronous o Asynchronous para ver el código correspondiente:

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)

Para obtener más información sobre cómo modificar la salida de su flujo de cambios, consulte la sección Modificar la salida del flujo de cambios en el manual de MongoDB Server.

Modificar watch() el comportamiento

El método watch() acepta parámetros opcionales, que representan opciones que se pueden usar para configurar la operación. Si no se especifica ninguna opción, el controlador no personaliza la operación.

La siguiente tabla describe las opciones que puede configurar para personalizar el comportamiento de 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': []}}

Información Adicional

Para obtener más información sobre los flujos de cambio, consulte Flujos de cambio en el manual de MongoDB Server.