versioned_collection.collection.tracking_collections.logs module

class versioned_collection.collection.tracking_collections.logs.LogTreeIdentifier

Bases: TypedDict

branch: str

version: int

class versioned_collection.collection.tracking_collections.logs.LogsCollection(*args: Any, **kwargs: Any)

Bases: _BaseTrackerCollection

Stores the logs for the tracked versions of the target collection.

This collection represents a log book for storing information about the registered versions of the target collection. A version has a number, a branch, a short description message and the timestamp at which it was registered.

The log is structured as a tree, each branch of the tree representing a versioning branch, therefore the log tree can be seen as a tree describing how the versions of the target collection have evolved in time, i.e., a tree that has as nodes the nodes of the version tree. Since this tree is a small data structure, it is cached in memory and automatically updated when new versions are registered, together with its persistent in-database representation.

class SCHEMA(version: int, branch: str, timestamp: datetime.datetime, message: str, prev: bson.ObjectId | None, next: List[bson.ObjectId])

Bases: object

branch: str

property id: bson.ObjectId | None

message: str

next: List[bson.ObjectId]

prev: bson.ObjectId | None

timestamp: datetime

version: int

weakly_equals(other: SCHEMA) → bool

Check if two log objects are weakly equal.

The weak equality does not take into account the linkage relation between log entries, i.e., it ignores the next and prev fields.

Parameters:

other – The object to check for weak equality.

Returns:

Whether the log entries are weakly equal.

add_log_entry(previous_version: int, previous_branch: str | None, current_branch: str, message: str, timestamp: datetime, with_id: bson.ObjectId | None = None) → Tuple[int, str]

Add a new entry to the log tree.

The log entries are created and added to the log tree when a new version of the target collection is registered. This method updates both the memory-cached tree and the persistent database tree.

Raises:

InvalidCollectionVersion – If the previous version and branch do not exist.

Parameters:

• previous_version – The version id of the previous version, i.e., the version which was modified to generate the version to be registered. Must be -1 only for the first version (root).

• previous_branch – The branch name of the previous version, i.e., the branch on which the previous version was registered.

• current_branch – The branch on which the new version should be registered.

• message – The message describing changes made to this version.

• timestamp – The time the version was registered.

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

Returns:

The version id and the branch name of the new entry.

build(message: str | None = None, timestamp: datetime | None = None, with_id: bson.ObjectId | None = None) → bool

Build this collection on the database.

Parameters:

• message – The message associated with the initial version of this collection.

• timestamp – The timestamp when this collection was created.

• with_id – The id of the root of the log tree.

Returns:

True if the collection was successfully built, False otherwise.

contains_version(version: int, branch: str) → bool

Return whether the given version is present in the log tree.

Parameters:

• version – The version id of the version.

• branch – The branch on which the version is registered.

Returns:

Whether the version exists in the log tree, i.e., if it is registered.

delete_subtree(version: Tuple[int, str]) → None

Delete the subtree of the version tree rooted in version.

Raises:

InvalidCollectionVersion – If version does not exist.

Parameters:

version – The root version of the subtree to delete.

get_log(branch: str, version: int | None = None, return_ids: bool = False) → List[SCHEMA]

Return the log sorted in descending order for the given branch.

If a version number is provided, the log will start at the collection version identified by version and branch, otherwise, the returned log will contain the entire history from the top of the branch to the root of the log tree.

Raises:

ValueError – If there is no node on the given branch, or if there is no node identified by version and branch.

Parameters:

• branch – The branch on which for which to retrieve the log.

• version – The version where to start the log from.

• return_ids – Whether to include the ids of the log documents as well.

Returns:

A list of log entries in descending order, i.e., the latest version at top.

get_log_doc_id(version: int, branch: str) → bson.ObjectId | None

Return the id of the document with the given version.

Parameters:

• version – The version of the log entry.

• branch – The branch of the log entry.

Returns:

The id of the log document in this collection, None if the version or the log tree does not exist.

get_log_entry(version: int, branch: str) → SCHEMA | None

Return the entry for the given version in the log tree.

Parameters:

• version – The version for which the entry will be retrieved.

• branch – The branch for which the entry will be retrieved.

Returns:

The entry in the log tree if it exists, None if the version or the log tree does not exist.

get_parent_branch(branch: str) → str | None

Retrieve the parent branch of the branch identified by branch.

Note

Branches are just unliked pointer documents. The branching structure of the collection is reflected in the log tree.

Raises:

• BranchNotFound – If no branch with the given name exists.

• InvalidCollectionState – If the given branch does not have a parent.

Parameters:

branch – The name of the branch whose parent should be returned.

Returns:

The parent branch, or None is branch='main'.

get_parent_version(version: Tuple[int, str]) → Tuple[int, str] | None

Return the version and branch of the previous version.

Raises:

InvalidCollectionVersion – If the version identified by version does not exist.

Parameters:

version – The version whose parent should be retrieved.

Returns:

The parent of the given version.

get_path_between_versions(current: Tuple[int, str], target: Tuple[int, str]) → Dict[Tuple[int, str], int]

Find the path between the given points in the log tree.

Note

The returned path includes the both ends.

Raises:

InvalidCollectionVersion – If the current or target versions do not exist in the log tree.

Parameters:

• current – The start point version.

• target – The end point version.

Returns:

An ordered dictionary representing the path that has to be followed in the log tree to get from the current version to the target version. The keys of the returned dictionary are tuples containing the version represented as (version, branch) and the values are the directions in time to be taken to get to the next version. The forward direction is represented as 1, and the backward direction as -1. The last entry, representing the target version, has the direction of the previous step as direction. If the two versions are on different branches with a common ancestor, then the direction at that node is 0.

get_previous_version_and_branch(current_version: int, current_branch: str) → Tuple[int, str] | None

Get the version and branch name of the previous version.

Looks up the current version in the log tree and returns the parent’s node version number and the branch name.

Raises:

InvalidCollectionVersion – If the given version is invalid or not registered.

Returns:

None if the collection is not tracked, the version and branch name of the previous node otherwise. If the current version is the root version, -1 will be returned as the previous version number.

get_versions_of_branch_tips(version: Tuple[int, str]) → List[Tuple[int, str]]

Get the versions of the leaves of the subtree rooted in version.

Parameters:

version – A version in the log tree.

Returns:

The versions of the tip of the branches of the log subtree rooted in version.

property log_tree: Tree | None

Return the log tree.

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

Move the versions starting at version to a new branch.

This updates the branch field of the log entries to be new_branch and resets the version counter field for version.

Raises:

ValueError – If version is the root version.

Parameters:

• version – The versions at which the new branch should start.

• new_branch – The name of the new branch.

reset() → bool

Reset this collection and the in-memory cache.

Returns:

True if the operation is successful, False otherwise.