Fluxos de alterações do MongoDB com Python

1
pymongo==3.8.0
2
dnspython

1
virtualenv venv  # sets up the environment
2
source venv/bin/activate  # activates the environment
3
pip3 install -r requirements.txt  # installs our dependencies

1
export CHANGE_STREAM_DB="mongodb+srv://user:<password>@example-xkfzv.mongodb.net/test?retryWrites=true"

1
import os
2
import pymongo
3
4
client = pymongo.MongoClient(os.environ['CHANGE_STREAM_DB'])
5
print(client.changestream.collection.insert_one({"hello": "world"}).inserted_id)

1
import os
2
import pymongo
3
from bson.json_util import dumps
4
5
client = pymongo.MongoClient(os.environ['CHANGE_STREAM_DB'])
6
change_stream = client.changestream.collection.watch()
7
for change in change_stream:
8
    print(dumps(change))
9
    print('') # for readability only

1
➜ python3 change_streams.py
2
{"_id": {"_data": {"$binary": "glsIjGUAAAABRmRfaWQAZFsIjGXiJuWPOIv2PgBaEAQIaEd7r8VFkazelcuRgfgeBA==", "$type": "00"}}, "operationType": "insert", "fullDocument": {"_id": {"$oid": "5b088c65e226e58f388bf63e"}, "hello": "world"}, "ns": {"db": "changestream", "coll": "collection"}, "documentKey": {"_id": {"$oid": "5b088c65e226e58f388bf63e"}}}

1
{
2
    _id: <resume_token>,
3
    operationType: "<type>",
4
    ns: {db: "<db name>", coll: "<collection name>"},
5
    documentKey: { <unique identifier> }
6
}

1
import os
2
import pymongo
3
4
client = pymongo.MongoClient(os.environ['CHANGE_STREAM_DB'])
5
client.changestream.collection.insert_one({"_id": 1, "hello": "world"})
6
client.changestream.collection.update_one({"_id": 1}, {"$set": {"hello": "mars"}})
7
client.changestream.collection.replace_one({"_id": 1} , {"bye": "world"})
8
client.changestream.collection.delete_one({"_id": 1})
9
client.changestream.collection.drop()

1
➜ python3 change_streams.py
2
{"fullDocument": {"_id": 1, "hello": "world"}, "documentKey": {"_id": 1}, "_id": {"_data": {"$binary": "glsIjuEAAAABRh5faWQAKwIAWhAECGhHe6/FRZGs3pXLkYH4HgQ=", "$type": "00"}}, "ns": {"coll": "collection", "db": "changestream"}, "operationType": "insert"}
3
4
{"documentKey": {"_id": 1}, "_id": {"_data": {"$binary": "glsIjuEAAAACRh5faWQAKwIAWhAECGhHe6/FRZGs3pXLkYH4HgQ=", "$type": "00"}}, "updateDescription": {"removedFields": [], "updatedFields": {"hello": "mars"}}, "ns": {"coll": "collection", "db": "changestream"}, "operationType": "update"}
5
6
{"fullDocument": {"bye": "world", "_id": 1}, "documentKey": {"_id": 1}, "_id": {"_data": {"$binary": "glsIjuEAAAADRh5faWQAKwIAWhAECGhHe6/FRZGs3pXLkYH4HgQ=", "$type": "00"}}, "ns": {"coll": "collection", "db": "changestream"}, "operationType": "replace"}
7
8
{"documentKey": {"_id": 1}, "_id": {"_data": {"$binary": "glsIjuEAAAAERh5faWQAKwIAWhAECGhHe6/FRZGs3pXLkYH4HgQ=", "$type": "00"}}, "ns": {"coll": "collection", "db": "changestream"}, "operationType": "delete"}
9
10
{"_id": {"_data": {"$binary": "glsIjuEAAAAFFFoQBAhoR3uvxUWRrN6Vy5GB+B4E", "$type": "00"}}, "operationType": "invalidate"}

1
import os
2
import pymongo
3
from bson.json_util import dumps
4
5
client = pymongo.MongoClient(os.environ['CHANGE_STREAM_DB'])
6
change_stream = client.changestream.collection.watch([{
7
    '$match': {
8
        'operationType': { '$in': ['insert'] }
9
    }
10
}])
11
12
for change in change_stream:
13
    print(dumps(change))
14
    print('')

1
"_id": {"_data": {"$binary": "glsIj84AAAACRh5faWQAKwIAWhAEvyfcy4djS8CUKRZ8tvWuOgQ=", "$type": "00"}}