change_stream – Watch changes on a collection, database, or cluster

Warning

This API is currently in beta, meaning the classes, methods, and behaviors described within may change before the full release. If you come across any bugs during your use of this API, please file a Jira ticket in the “Python Driver” project at https://jira.mongodb.org/browse/PYTHON.

Watch changes on a collection, a database, or the entire cluster.

class pymongo.asynchronous.change_stream.AsyncChangeStream(target, pipeline, full_document, resume_after, max_await_time_ms, batch_size, collation, start_at_operation_time, session, start_after, comment=None, full_document_before_change=None, show_expanded_events=None)

The internal abstract base class for change stream cursors.

Should not be called directly by application developers. Use pymongo.asynchronous.collection.AsyncCollection.watch(), pymongo.asynchronous.database.AsyncDatabase.watch(), or pymongo.asynchronous.mongo_client.AsyncMongoClient.watch() instead.

Added in version 3.6.

See also

The MongoDB documentation on changeStreams.

Parameters:

  • target (Union[AsyncMongoClient[_DocumentType], AsyncDatabase[_DocumentType], AsyncCollection[_DocumentType]])

  • pipeline (Optional[_Pipeline])

  • full_document (Optional[str])

  • resume_after (Optional[Mapping[str, Any]])

  • max_await_time_ms (Optional[int])

  • batch_size (Optional[int])

  • collation (Optional[_CollationIn])

  • start_at_operation_time (Optional[Timestamp])

  • session (Optional[AsyncClientSession])

  • start_after (Optional[Mapping[str, Any]])

  • comment (Optional[Any])

  • full_document_before_change (Optional[str])

  • show_expanded_events (Optional[bool])

property alive: bool

Does this cursor have the potential to return more data?

Note

Even if alive is True, next() can raise StopIteration and try_next() can return None.

Added in version 3.8.

async close()

Close this AsyncChangeStream.

Return type:

None

async next()

Advance the cursor.

This method blocks until the next change document is returned or an unrecoverable error is raised. This method is used when iterating over all changes in the cursor. For example:

try:
    resume_token = None
    pipeline = [{'$match': {'operationType': 'insert'}}]
    async with await db.collection.watch(pipeline) as stream:
        async for insert_change in stream:
            print(insert_change)
            resume_token = stream.resume_token
except pymongo.errors.PyMongoError:
    # The AsyncChangeStream encountered an unrecoverable error or the
    # resume attempt failed to recreate the cursor.
    if resume_token is None:
        # There is no usable resume token because there was a
        # failure during AsyncChangeStream initialization.
        logging.error('...')
    else:
        # Use the interrupted AsyncChangeStream's resume token to create
        # a new AsyncChangeStream. The new stream will continue from the
        # last seen insert change without missing any events.
        async with await db.collection.watch(
                pipeline, resume_after=resume_token) as stream:
            async for insert_change in stream:
                print(insert_change)

Raises StopIteration if this AsyncChangeStream is closed.

Return type:

_DocumentType

property resume_token: Mapping[str, Any] | None

The cached resume token that will be used to resume after the most recently returned change.

Added in version 3.9.

async try_next()

Advance the cursor without blocking indefinitely.

This method returns the next change document without waiting indefinitely for the next change. For example:

async with await db.collection.watch() as stream:
    while stream.alive:
        change = await stream.try_next()
        # Note that the AsyncChangeStream's resume token may be updated
        # even when no changes are returned.
        print("Current resume token: %r" % (stream.resume_token,))
        if change is not None:
            print("Change document: %r" % (change,))
            continue
        # We end up here when there are no recent changes.
        # Sleep for a while before trying again to avoid flooding
        # the server with getMore requests when no changes are
        # available.
        asyncio.sleep(10)

If no change document is cached locally then this method runs a single getMore command. If the getMore yields any documents, the next document is returned, otherwise, if the getMore returns no documents (because there have been no changes) then None is returned.

Returns:

The next change document or None when no document is available after running a single getMore or when the cursor is closed.

Return type:

_DocumentType | None

Added in version 3.8.

class pymongo.asynchronous.change_stream.AsyncClusterChangeStream(target, pipeline, full_document, resume_after, max_await_time_ms, batch_size, collation, start_at_operation_time, session, start_after, comment=None, full_document_before_change=None, show_expanded_events=None)

A change stream that watches changes on all collections in the cluster.

Should not be called directly by application developers. Use helper method pymongo.asynchronous.mongo_client.AsyncMongoClient.watch() instead.

Added in version 3.7.

Parameters:

  • target (Union[AsyncMongoClient[_DocumentType], AsyncDatabase[_DocumentType], AsyncCollection[_DocumentType]])

  • pipeline (Optional[_Pipeline])

  • full_document (Optional[str])

  • resume_after (Optional[Mapping[str, Any]])

  • max_await_time_ms (Optional[int])

  • batch_size (Optional[int])

  • collation (Optional[_CollationIn])

  • start_at_operation_time (Optional[Timestamp])

  • session (Optional[AsyncClientSession])

  • start_after (Optional[Mapping[str, Any]])

  • comment (Optional[Any])

  • full_document_before_change (Optional[str])

  • show_expanded_events (Optional[bool])

class pymongo.asynchronous.change_stream.AsyncCollectionChangeStream(target, pipeline, full_document, resume_after, max_await_time_ms, batch_size, collation, start_at_operation_time, session, start_after, comment=None, full_document_before_change=None, show_expanded_events=None)

A change stream that watches changes on a single collection.

Should not be called directly by application developers. Use helper method pymongo.asynchronous.collection.AsyncCollection.watch() instead.

Added in version 3.7.

Parameters:

  • target (Union[AsyncMongoClient[_DocumentType], AsyncDatabase[_DocumentType], AsyncCollection[_DocumentType]])

  • pipeline (Optional[_Pipeline])

  • full_document (Optional[str])

  • resume_after (Optional[Mapping[str, Any]])

  • max_await_time_ms (Optional[int])

  • batch_size (Optional[int])

  • collation (Optional[_CollationIn])

  • start_at_operation_time (Optional[Timestamp])

  • session (Optional[AsyncClientSession])

  • start_after (Optional[Mapping[str, Any]])

  • comment (Optional[Any])

  • full_document_before_change (Optional[str])

  • show_expanded_events (Optional[bool])

class pymongo.asynchronous.change_stream.AsyncDatabaseChangeStream(target, pipeline, full_document, resume_after, max_await_time_ms, batch_size, collation, start_at_operation_time, session, start_after, comment=None, full_document_before_change=None, show_expanded_events=None)

A change stream that watches changes on all collections in a database.

Should not be called directly by application developers. Use helper method pymongo.asynchronous.database.AsyncDatabase.watch() instead.

Added in version 3.7.

Parameters:

  • target (Union[AsyncMongoClient[_DocumentType], AsyncDatabase[_DocumentType], AsyncCollection[_DocumentType]])

  • pipeline (Optional[_Pipeline])

  • full_document (Optional[str])

  • resume_after (Optional[Mapping[str, Any]])

  • max_await_time_ms (Optional[int])

  • batch_size (Optional[int])

  • collation (Optional[_CollationIn])

  • start_at_operation_time (Optional[Timestamp])

  • session (Optional[AsyncClientSession])

  • start_after (Optional[Mapping[str, Any]])

  • comment (Optional[Any])

  • full_document_before_change (Optional[str])

  • show_expanded_events (Optional[bool])