versioned_collection.collection.tracking_collections.deltas module

class versioned_collection.collection.tracking_collections.deltas.DeltasCollection(*args: Any, **kwargs: Any)

Bases: _BaseTrackerCollection

Stores the deltas between different versions of the target collection.

Each delta document in this collection reflects the changes and therefore, the actions needed to be performed to move between different versions of the documents in the target collection. If no deltas for a particular document are present, then that document has not been modified since the target collection was initialised for versioning.

The deltas documents store information about the changes of particular documents in the target collection for a specific version and a specific branch of the versioning system. Even though the deltas for all documents are stored together, they can be grouped by the document in the target collection that they modify. The per-document deltas are tree structures, where the branches of the tree are a subset of the branches of the version tree (because some documents may not be modified on some branches). The per-document delta tree allows applying changes in both directions of time for a specific document. Since the target collection may contain multiple documents, this collection represents a set of per-document delta trees.

To get to a specific version of the target collection we need the path in version tree or log tree between the two versions. All the deltas between the start and the end points of the path will be retrieved according to the direction in which we should navigate the delta trees. The delta trees for each modified document between the two versions are built and intersected with the given path, to get the sequence of deltas that have to be applied to a document to move across versions.

class SCHEMA(document_id: Any, collection_version_id: int, branch: str, timestamp: datetime.datetime, forward: bytes, backward: bytes, prev: bson.ObjectId | None, next: List[bson.ObjectId])

Bases: object

backward: bytes

branch: str

collection_version_id: int

document_id: Any

forward: bytes

next: List[bson.ObjectId]

prev: bson.ObjectId | None

timestamp: datetime

add_delta(document_old: Dict[str, Any], document_new: Dict[str, Any], document_id: bson.ObjectId, collection_version: int, branch: str, timestamp: datetime, branch_history: List[Tuple[int, str]], with_id: bson.ObjectId | None = None) → bson.ObjectId | None

Compute and records the deltas between the given document versions.

Raises:

InvalidCollectionState – If some deltas for the current collection version cannot be identified.

Parameters:

• document_old – The old version of the document.

• document_new – The new version of the document, that contains changes.

• document_id – The id of the modified document.

• collection_version – The version of the tracked collection to which the changes to the given document should be registered.

• branch – The branch that the modified target document belongs to.

• timestamp – The date and time when the delta was registered.

• branch_history – A set containing (version, branch) tuples from the previous version to the root of the version tree.

• with_id – An optional ObjectId used for inserting the new entry into this collection. This is used when adding deltas from a local to a remote collection.

Returns:

The id of the delta document, or None if the two versions of the document are unchanged.

apply_deltas(per_document_deltas: Dict[Any, List[deepdiff.Delta]], documents: List[Dict[str, Any]], return_current_documents: bool = False) → Dict[Any, Dict[str, Any]] | Tuple[Dict[Any, Dict[str, Any]], Dict[Any, Dict[str, Any]]]

Update the given documents and return them.

Applies the deltas between two versions of the target collection. It uses the given deltas grouped by document and sorted by the direction on which they have to be applied and sequentially updates each document to get to the version of the document for the target version of the tracked collection.

Parameters:

• per_document_deltas – The prefetched and sorted list of deltas that have to be applied, grouped by document id.

• documents – The list of documents that will be updated.

• return_current_documents – Whether to return the current documents grouped by id.

Param:

If return_current_documents is set, the current documents are returned as well.

Returns:

The updated documents grouped by their '_id' field. If The documents that are empty should be removed from the target collection.

build() → bool

Create the collection on the database.

Returns:

False if the collection already exists, True otherwise.

delete_subtrees(root: Tuple[int, str], leaves: List[Tuple[int, str]]) → None

Delete the deltas registered after a specific version.

Parameters:

• root – The root of the subtree of the version tree from which deltas will be removed.

• leaves – The versions of the leaves of the version subtree.

get_delta_documents_in_path(path: Dict[Tuple[int, str], int], sorting_order: int | None = None) → pymongo.command_cursor.CommandCursor

Get the delta documents grouped by tracked document’s id in path.

Parameters:

• path – The path in the version tree from which to pull the delta documents.

• sorting_order – The order in which to sort the delta documents by timestamp. 1 means ascending and -1 means descending. If omitted, the sorting step is skipped.

Returns:

The delta documents.

get_deltas(path: Dict[Tuple[int, str], int]) → Dict[Any, List[deepdiff.Delta]]

Retrieve the deltas across the given path of versions.

Parameters:

path – The path between two versions. The keys identify the version (i.e., (version, branch) tuples) and the values the direction in time to take to move between versions.

Returns:

A list of tuples containing the requested deltas, grouped by the documents’ id.

insert_delta_docs(delta_docs: List[Dict[str, Any]]) → None

Insert a list of delta documents into this collection.

Warning

This method modifies the delta documents and removes the ids of the documents from the next field that are not part of the given delta_docs list.

This is used during remote-local synchronisation of a branch, therefore the inserted deltas are slightly modified versions of the local deltas. Since a single branch can be pushed or pulled at a time, the forward references to other delta documents that are not in the given list are removed. Also, the parent deltas are updated to include the first delta document in delta_docs in its forward references field, i.e., next.

Parameters:

delta_docs – The delta documents to be inserted. It is assumed that the deltas are sorted.

rebranch(start_version: Tuple[int, str], new_branch: str, num_versions: int) → None

Move the deltas after start_version to another branch.

Parameters:

• start_version – The version which should be moved to a new branch.

• new_branch – The name of the new branch.

• num_versions – The number of versions to move, i.e., the length of the branch starting at start_version.