Monitore dados com change streams

Visão geral

Neste guia, você pode aprender como usar um fluxo de alterações para monitorar alterações em tempo real em seu banco de dados. Um change stream é um recurso do MongoDB Server que permite que seu aplicativo se inscreva em alterações de dados em uma collection, banco de dados ou sistema.

Abrir um fluxo de alterações

Para abrir um fluxo de alteração, chame o método watch() . A instância na qual você chama o método watch() determina o escopo de eventos que o change stream escuta. Você pode chamar o método watch() nas seguintes classes:

• MongoClient: Para monitorar todas as alterações no sistema MongoDB

• Database: Para monitorar alterações em todas as coleções no banco de dados

• Collection: Para monitorar alterações na coleção

O exemplo a seguir abre um fluxo de alteração na coleção restaurants e gera alterações à medida que ocorrem. Selecione a aba Synchronous ou Asynchronous para ver o código correspondente:

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 começar a observar as alterações, execute o aplicação. Em seguida, em um aplicação ou shell separado, modifique a coleção restaurants. O exemplo a seguir atualiza um documento com um valor de campo name de Blarney Castle. Selecione a aba Synchronous ou Asynchronous para ver o código correspondente:

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)

Quando você atualiza a coleção, o aplicativo de fluxo de alterações imprime a alteração conforme ela ocorre. O evento de alteração impresso é semelhante ao seguinte:

{'_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 a saída change stream

Você pode passar o parâmetro pipeline para o método watch() para modificar a saída do change stream. Esse parâmetro permite que você observe somente eventos de alteração especificados. Formate o parâmetro como uma lista de objetos que cada representa um estágio de agregação.

Você pode especificar os seguintes estágios no parâmetro pipeline :

• $addFields

• $match

• $project

• $replaceRoot

• $replaceWith

• $redact

• $set

• $unset

O exemplo a seguir usa o parâmetro pipeline para abrir um fluxo de alterações que registra somente operações de atualização. Selecione a aba Synchronous ou Asynchronous para ver o código correspondente:

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 saber mais sobre como modificar a saída do change stream, consulte a seção Modificar a saída do change stream no manual do MongoDB Server .

Modificar Comportamento do watch()

O método watch() aceita parâmetros opcionais, que representam opções que você pode utilizar para configurar a operação. Se você não especificar nenhuma opção, o driver não personalizará a operação.

A tabela seguinte descreve as opções que você pode definir para personalizar o comportamento 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': []}}

Informações adicionais

Para saber mais sobre fluxos de alterações, consulte Change Streams de alterações no manual do MongoDB Server .