swh.storage.utils module#
- swh.storage.utils.map_optional(f: Callable[[T1], T2], x: T1 | None) T2 | None[source]#
Execute function f on x if x is not None and return the result. Return None otherwise.
- Parameters:
f – Function to be called on non-None value
x – The value to apply f onto (or None)
- Returns:
f(x) if x is not None else None.
- swh.storage.utils.get_partition_bounds_bytes(i: int, n: int, nb_bytes: int) Tuple[bytes, bytes | None][source]#
Splits the range [0; 2^(nb_bytes*8)) into n same-length intervals, and returns the boundaries of this interval (both inclusive); or None as upper bound, if this is the last partition
n must be a power of 2.
>>> get_partition_bounds_bytes(0, 16, 2) == (b'\x00\x00', b'\x10\x00') True >>> get_partition_bounds_bytes(1, 16, 2) == (b'\x10\x00', b'\x20\x00') True >>> get_partition_bounds_bytes(14, 16, 2) == (b'\xe0\x00', b'\xf0\x00') True >>> get_partition_bounds_bytes(15, 16, 2) == (b'\xf0\x00', None) True
- swh.storage.utils.extract_collision_hash(error_message: str) Tuple[str, bytes] | None[source]#
Utilities to extract the hash information from a hash collision error.
Hash collision error message are of the form: ‘Key (<hash-type>)=(<double-escaped-hash) already exists.’
for example: ‘Key (sha1)=(x34973274ccef6ab4dfaaf86599792fa9c3fe4689) already exists.’
- Returns:
A formatted string
- swh.storage.utils.content_hex_hashes(content: Dict[str, bytes]) Dict[str, str][source]#
Convert bytes hashes into hex hashes.
- swh.storage.utils.content_bytes_hashes(content: Dict[str, str]) Dict[str, bytes][source]#
Convert bytes hashes into hex hashes.