中文
devexamples

PyMongo Change Stream Resume Token

Watch MongoDB collection changes with PyMongo, persisting resume tokens so listeners recover exactly where they left off after restart.

PythonData Fetching

Code

snippet
import json

from pymongo import MongoClient
from pymongo.collection import Collection

client = MongoClient('mongodb://localhost:27017', serverSelectionTimeoutMS=3000)
orders: Collection = client['shop']['orders']


def load_token() -> dict | None:
    try:
        with open('.resume-token.json') as fh:
            return json.load(fh)
    except FileNotFoundError:
        return None


def save_token(token: dict) -> None:
    with open('.resume-token.json', 'w') as fh:
        json.dump(token, fh)


def listen(max_events: int = 5) -> list[dict]:
    events: list[dict] = []
    pipeline = [{'$match': {'operationType': {'$in': ['insert', 'replace']}}}]

    with orders.watch(
        pipeline, full_document='updateLookup', resume_after=load_token()
    ) as stream:
        for event in stream:
            doc = event.get('fullDocument', {})
            events.append({
                'op': event['operationType'],
                'id': str(doc.get('_id', '')),
                'status': doc.get('status', ''),
            })
            save_token(stream.resume_token)
            if len(events) >= max_events:
                break
    return events


orders.insert_one({'status': 'created', 'total': 99})
print('captured events:', listen(max_events=1))

Related Examples