swh.storage.algos.snapshot module#

swh.storage.algos.snapshot.snapshot_get_all_branches(storage: StorageInterface, snapshot_id: bytes) → Optional[Snapshot][source]#

Get all the branches for a given snapshot

Parameters:
Returns:

A snapshot objects populated with all known branches if the snapshot is found or None.

swh.storage.algos.snapshot.snapshot_get_latest(storage: StorageInterface, origin: str, allowed_statuses: Optional[List[str]] = None, branches_count: Optional[int] = None) → Optional[Snapshot][source]#

Get the latest snapshot for the given origin, optionally only from visits that have one of the given allowed_statuses.

The branches of the snapshot are iterated in the lexicographical order of their names.

Parameters:

  • storage – Storage instance

  • origin – the origin’s URL

  • allowed_statuses – list of visit statuses considered to find the latest snapshot for the visit. For instance, allowed_statuses=['full'] will only consider visits that have successfully run to completion.

  • branches_count – Optional parameter to retrieve snapshot with all branches (default behavior when None) or not. If set to positive number, the snapshot will be partial with only that number of branches.

Raises:

ValueError if branches_count is not a positive value –

Returns:

The snapshot object if one is found matching the criteria or None.

swh.storage.algos.snapshot.snapshot_id_get_from_revision(storage: StorageInterface, origin: str, revision_id: bytes) → Optional[bytes][source]#

Retrieve the most recent snapshot id targeting the revision_id for the given origin.

Warning This is a potentially highly costly operation

Returns

The snapshot id if found. None otherwise.

swh.storage.algos.snapshot.visits_and_snapshots_get_from_revision(storage: StorageInterface, origin: str, revision_id: bytes) → Iterator[Tuple[OriginVisit, OriginVisitStatus, Snapshot]][source]#

Retrieve all visits, visit statuses, and matching snapshot of the given origin, such that the snapshot targets the revision_id.

Warning This is a potentially highly costly operation

Yields:

Tuples of (visit, status, snapshot)

swh.storage.algos.snapshot.snapshot_resolve_branch_target(storage: StorageInterface, snapshot_id: bytes, branch_obj: Optional[SnapshotBranch], max_length: int = 100) → Tuple[Optional[SnapshotBranch], List[bytes]][source]#

Return the final target of a snapshot branch along with its resolve chain. Return ‘None’ as the target in case the chain is longer than the given max_length or has cycles

Parameters:

  • storage – Storage instance

  • snapshot_id – snapshot identifier

  • branch_obj – snapshot branch object to resolve

  • max_length – maximum chain length before breaking

Returns:

The first branch that isn’t an alias and the resolve chain

swh.storage.algos.snapshot.snapshot_resolve_alias(storage: StorageInterface, snapshot_id: bytes, alias_name: bytes) → Optional[SnapshotBranch][source]#

Transitively resolve snapshot branch alias to its real target, and return it; ie. follows every branch that is an alias, until a branch that isn’t an alias is found.

Parameters:

  • storage – Storage instance

  • snapshot_id – snapshot identifier

  • alias_name – name of the branch alias to resolve

Returns:

The first branch that isn’t an alias, in the alias chain; or None if there is no such branch (ie. either because of a cycle alias, or a dangling branch).