neosqlite.changestream module

neosqlite.changestream._registry_entry(db: Any) → dict[str, Any]

Per-connection registry of active ChangeStreams (#110).

Keyed by id(db) with a strong reference to db kept in the entry, which pins the id for as long as any stream is registered.

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)

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)

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

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()

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()

Drop shared triggers when the LAST stream for this collection closes, and purge its now-unconsumable events (#110, #111).

Called from close(); the _closed flag is set by the caller.

_advance_watermark(change_id: int) → None

_prune_consumed() → None

Delete events that EVERY active stream on this collection has already consumed (#111). With no co-streams this is just this stream’s own watermark.

close() → None

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.