Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/utils.py
# Copyright (C) 2019-2020 The Software Heritage developers | # Copyright (C) 2019-2020 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import re | import re | ||||
from typing import Optional, Tuple | from typing import Dict, Optional, Tuple | ||||
from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import ( | ||||
hash_to_bytes, hash_to_hex, DEFAULT_ALGORITHMS | |||||
) | |||||
def _is_power_of_two(n: int) -> bool: | def _is_power_of_two(n: int) -> bool: | ||||
return n > 0 and n & (n-1) == 0 | return n > 0 and n & (n-1) == 0 | ||||
def get_partition_bounds_bytes( | def get_partition_bounds_bytes( | ||||
i: int, n: int, nb_bytes: int) -> Tuple[bytes, Optional[bytes]]: | i: int, n: int, nb_bytes: int) -> Tuple[bytes, Optional[bytes]]: | ||||
▲ Show 20 Lines • Show All 42 Lines • ▼ Show 20 Lines | def extract_collision_hash(error_message: str) -> Optional[Tuple[str, bytes]]: | ||||
""" | """ | ||||
pattern = r'\w* \((?P<type>[^)]+)\)=\(\\x(?P<id>[a-f0-9]+)\) \w*' | pattern = r'\w* \((?P<type>[^)]+)\)=\(\\x(?P<id>[a-f0-9]+)\) \w*' | ||||
result = re.match(pattern, error_message) | result = re.match(pattern, error_message) | ||||
if result: | if result: | ||||
hash_type = result.group('type') | hash_type = result.group('type') | ||||
hash_id = result.group('id') | hash_id = result.group('id') | ||||
return hash_type, hash_to_bytes(hash_id) | return hash_type, hash_to_bytes(hash_id) | ||||
return None | return None | ||||
def content_hex_hashes(content: Dict[str, bytes]) -> Dict[str, str]: | |||||
"""Convert bytes hashes into hex hashes. | |||||
""" | |||||
return { | |||||
algo: hash_to_hex(content[algo]) for algo in DEFAULT_ALGORITHMS | |||||
} | |||||
def content_bytes_hashes(content: Dict[str, str]) -> Dict[str, bytes]: | |||||
"""Convert bytes hashes into hex hashes. | |||||
""" | |||||
return { | |||||
algo: hash_to_bytes(content[algo]) for algo in DEFAULT_ALGORITHMS | |||||
} |