neosqlite.changestream module

class neosqlite.changestream.ChangeStream(collection: Collection, pipeline: list[dict[str, Any]] | None = None, full_document: str | None = None, resume_after: dict[str, Any] | None = None, max_await_time_ms: int | None = None, batch_size: int | None = None, collation: dict[str, Any] | None = None, start_at_operation_time: Any | None = None, session: ClientSession | None = None, start_after: dict[str, Any] | None = None)[source]

Bases: object

A change stream that watches for changes on a collection.

This implementation uses SQLite’s built-in features to monitor changes. It provides an iterator interface to receive change events.

__init__(collection: Collection, pipeline: list[dict[str, Any]] | None = None, full_document: str | None = None, resume_after: dict[str, Any] | None = None, max_await_time_ms: int | None = None, batch_size: int | None = None, collation: dict[str, Any] | None = None, start_at_operation_time: Any | None = None, session: ClientSession | None = None, start_after: dict[str, Any] | None = None)[source]

Initialize a change stream for a specific collection.

Parameters:

  • collection (Collection) – The collection to monitor for changes.

  • pipeline (list[dict[str, Any]], optional) – A pipeline of operations to apply to the change stream.

  • full_document (str, optional) – Specifies whether to include the full document in change events.

  • resume_after (dict[str, Any], optional) – A resume token to start the change stream from a specific point.

  • max_await_time_ms (int, optional) – The maximum time in milliseconds to wait for change events.

  • batch_size (int, optional) – The batch size for the change stream.

  • collation (dict[str, Any], optional) – Collation options to apply to change events.

  • start_at_operation_time (Any, optional) – Operation time to start the change stream from.

  • session (Any, optional) – The session to use for the change stream.

  • start_after (dict[str, Any], optional) – A document ID to start the change stream from.

static _sanitize_collection_name(name: str) → str[source]

Validate and sanitize a collection name to prevent SQL injection.

Parameters:

name – The collection name to validate.

Returns:

The validated collection name.

Raises:

ValueError – If the collection name contains invalid characters.

_setup_triggers()[source]

Set up SQLite triggers to capture changes to the collection.

This method ensures that triggers are created in the SQLite database to log INSERT, UPDATE, and DELETE operations on the specified collection. These triggers insert records into a change tracking table, enabling the change stream to monitor these events.

Triggers are created dynamically using SQL commands. They are designed to capture the essential details of each change operation, including the operation type, document ID, and data.

_cleanup_triggers()[source]

Clean up the triggers when the change stream is closed.

This method ensures that triggers created for capturing changes to the collection are properly dropped from the SQLite database when the change stream is no longer needed. This cleanup helps in freeing up resources and avoiding unnecessary logging.

The method handles exceptions gracefully, ensuring that any errors during the cleanup process are ignored, thus allowing the change stream to close without interruption.

close() → None[source]

Close the change stream and clean up resources.

This method ensures that the change stream is properly closed and resources are released. It sets the _closed flag to True and calls the _cleanup_triggers method to clean up any triggers that were set up for capturing changes to the collection. This helps in freeing up resources and avoiding unnecessary logging or data handling.

By calling this method, the change stream is effectively terminated, and no further change events will be received.