Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/in_memory.py
Show First 20 Lines • Show All 339 Lines • ▼ Show 20 Lines | def content_find(self, content: Dict[str, Any]) -> List[Content]: | ||||
found.append(self._content_indexes[algo][hash]) | found.append(self._content_indexes[algo][hash]) | ||||
if not found: | if not found: | ||||
return [] | return [] | ||||
keys = list(set.intersection(*found)) | keys = list(set.intersection(*found)) | ||||
return [self._contents[key] for key in keys] | return [self._contents[key] for key in keys] | ||||
def content_missing(self, content, key_hash="sha1"): | def content_missing( | ||||
for cont in content: | self, contents: List[Dict[str, Any]], key_hash: str = "sha1" | ||||
for (algo, hash_) in cont.items(): | ) -> Iterable[bytes]: | ||||
if key_hash not in DEFAULT_ALGORITHMS: | |||||
raise StorageArgumentException( | |||||
"key_hash should be one of {','.join(DEFAULT_ALGORITHMS)}" | |||||
) | |||||
for content in contents: | |||||
for (algo, hash_) in content.items(): | |||||
if algo not in DEFAULT_ALGORITHMS: | if algo not in DEFAULT_ALGORITHMS: | ||||
continue | continue | ||||
if hash_ not in self._content_indexes.get(algo, []): | if hash_ not in self._content_indexes.get(algo, []): | ||||
yield cont[key_hash] | yield content[key_hash] | ||||
break | break | ||||
def content_missing_per_sha1(self, contents): | def content_missing_per_sha1(self, contents): | ||||
for content in contents: | for content in contents: | ||||
if content not in self._content_indexes["sha1"]: | if content not in self._content_indexes["sha1"]: | ||||
yield content | yield content | ||||
def content_missing_per_sha1_git(self, contents): | def content_missing_per_sha1_git(self, contents): | ||||
▲ Show 20 Lines • Show All 910 Lines • Show Last 20 Lines |