change_stream_reader.py

'''
Script for picking up changes from the change stream
'''
import os
import json
import pymongo
from bson.json_util import dumps
from bson import ObjectId

client = pymongo.MongoClient('mongodb://localhost:27017/')
db = client['test']

def handle_event_expiry(id):
	print(id)
	events = db.notifications.find({'event_id': ObjectId(id)})

	for i in events:
	    print(f'processing events {i}')
    


change_stream = db.expiry_event.watch()

for change in change_stream:

    event = json.loads(dumps(change))
    if event.get('operationType') == 'delete':
        expired_event = event.get('documentKey', {}).get('_id', {}).get('$oid')
        handle_event_expiry(expired_event)