Skip to main content

Mongo Observer

Mongo Observer aims to provide an easy way to asynchronously subscribe to state change events on a given mongo collection.

Installation

pip install mongo_observer

Usage

The Observer

The Observer handles state change observation on a given collection and dispatches events, delegating the responsibility to a handler.

Handlers

Handlers are objects that implement the OperationHandler protocol and implement the following abstract methods:

async def on_insert(self, operation: Dict[str, Any])

Where operation is a dict containing a document corresponding to an operation on oplog. It will contain the following keys:

  • ts: Timestamp of the operationa

  • h: An unique signed long identifier of the operation

  • op: A character representing the type of the operation

  • ns: A namespace string formed with the concatenation of ‘database.collection’

  • o: The inserted document

async def on_update(self, operation: Dict[str, Any])

Where operation is a dict containing a document corresponding to an operation on oplog. It will contain the following keys:

  • ts: Timestamp of the operation

  • h: An unique signed long identifier of the operation

  • op: A character representing the type of the operation

  • ns: A namespace string formed with the concatenation of ‘database.collection’

  • o: The operation data performed on the document

  • o2: A dict with a single _id key of the document to be updated

async def on_delete(self, operation: Dict[str, Any])

Where operation is a dict containing a document corresponding to an operation on oplog. It will contain the following keys:

  • ts: Timestamp of the operation

  • h: An unique signed long identifier of the operation

  • op: A character representing the type of the operation

  • ns: A namespace string formed with the concatenation of ‘database.collection’

  • o: A dict with a single _id key, of the ddeleted document

ReactiveCollection

A ReactiveCollection is a read-only, in-memory, non-persistent replica of a remote mongo collection. It reacts to state changes caused by write operations (inserts, updates and deletes) on the remote collection.

import asyncio

from motor.motor_asyncio import AsyncIOMotorClient

from mongo_observer.observer import Observer
from mongo_observer.operation_handlers import ReactiveCollection


async def run(loop):
    client = AsyncIOMotorClient('mongodb://127.0.0.1')

    collection_to_observe = client['your_db']['your_collection']
    reactive_collection = await ReactiveCollection.init_async(collection_to_observe)

    observer = await Observer.init_async(oplog=client['local']['oplog.rs'],
                                         operation_handler=reactive_collection,
                                         namespace_filter='your_db.your_collection')

    loop.create_task(observer.observe_changes())


loop = asyncio.get_event_loop()
loop.run_until_complete(run(loop))
loop.run_forever()

Once ReactiveCollection.init_async runs, it will perform an initial sync with the remote collection. We then connect it as the operation_handler of an Observer, and we are ready to observe_changes to the current state of the collection.

ReactivePartialCollection

to-do

Release files for mongo_observer 0.0.1

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for mongo_observer 0.0.1
File Size Uploaded
mongo_observer-0.0.1.tar.gz 4.9 kB Details

Release files / mongo_observer-0.0.1.tar.gz

Download URL mongo_observer-0.0.1.tar.gz
Size 4.9 kB
Tags Source
SHA-256 checksum
How to use checksums
bf948966477feb8e73364a5d47862f96ece3ca0e6955778a7f07ae1c0e13be8a
BLAKE2b-256 checksum
How to use checksums
b917b13d50987b88b4520624514b305acfa9670161767086aa5fd94303991115
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No

Release history Release notifications | RSS feed

This release

0.0.1 This release

1 release file

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page