swh.storage.utils module

swh.storage.utils.now() datetime.datetime[source]
swh.storage.utils.map_optional(f: Callable[[swh.storage.utils.T1], swh.storage.utils.T2], x: Optional[swh.storage.utils.T1]) Optional[swh.storage.utils.T2][source]
swh.storage.utils.get_partition_bounds_bytes(i: int, n: int, nb_bytes: int) Tuple[bytes, Optional[bytes]][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) Optional[Tuple[str, bytes]][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.

swh.storage.utils.remove_keys(d: Dict[swh.storage.utils.T1, swh.storage.utils.T2], keys: Tuple[swh.storage.utils.T1, ...]) Dict[swh.storage.utils.T1, swh.storage.utils.T2][source]

Returns a copy of d minus the given keys.

swh.storage.utils.round_to_milliseconds(date)[source]

Round datetime to milliseconds before insertion, so equality doesn’t fail after a round-trip through a DB (eg. Cassandra)