Supervise los datos con Change Streams

Overview

En esta guía, puedes aprender a utilizar un flujo de cambios para supervisar los cambios en tiempo real de tu base de datos. Un flujo de cambios es una funcionalidad de MongoDB Server que permite que la aplicación se suscriba a 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, llama al método watch(). La instancia en la que se llama al método watch() determina el alcance de los eventos para los que el flujo de cambios escucha. Puedes llamar al método watch() en las siguientes clases:

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

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

• Collection: Para supervisar los cambios en la colección

El siguiente ejemplo abre un flujo de cambios en la colección restaurants y muestra los cambios a medida que ocurren. Selecciona la pestaña Synchronous o 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 comenzar a observar los cambios, ejecuta la aplicación. Luego, en una aplicación o shell aparte, modifica la colección restaurants. El siguiente ejemplo actualiza un documento con un valor de campo name de Blarney Castle. Selecciona 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)

Cuando actualizas la colección, la aplicación del flujo de cambios imprime el cambio a medida que ocurre. 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 Cambio de Flujo

Puedes pasar el parámetro pipeline al método watch() para modificar la salida del flujo de cambios. Este parámetro te permite monitorear solo los eventos de cambio especificados. Da formato al 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 solo 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 tu flujo de cambios, consulta la sección Modificar salida del flujo de cambio en el manual del MongoDB Server.

Modificar el comportamiento de watch()

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

La siguiente tabla describe las opciones que puedes 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 saber más sobre las change streams, consulta Change Streams en el manual de MongoDB Server.