Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/filter.py
Show All 31 Lines | class FilteringProxyStorage: | ||||
def __init__(self, storage): | def __init__(self, storage): | ||||
self.storage: StorageInterface = get_storage(**storage) | self.storage: StorageInterface = get_storage(**storage) | ||||
def __getattr__(self, key): | def __getattr__(self, key): | ||||
if key == "storage": | if key == "storage": | ||||
raise AttributeError(key) | raise AttributeError(key) | ||||
return getattr(self.storage, key) | return getattr(self.storage, key) | ||||
def content_add(self, content: List[Content]) -> Dict: | def content_add(self, content: List[Content]) -> Dict[str, int]: | ||||
contents_to_add = self._filter_missing_contents(content) | contents_to_add = self._filter_missing_contents(content) | ||||
return self.storage.content_add( | return self.storage.content_add( | ||||
[x for x in content if x.sha256 in contents_to_add] | [x for x in content if x.sha256 in contents_to_add] | ||||
) | ) | ||||
def skipped_content_add(self, content: List[SkippedContent]) -> Dict: | def skipped_content_add(self, content: List[SkippedContent]) -> Dict[str, int]: | ||||
contents_to_add = self._filter_missing_skipped_contents(content) | contents_to_add = self._filter_missing_skipped_contents(content) | ||||
return self.storage.skipped_content_add( | return self.storage.skipped_content_add( | ||||
[x for x in content if x.sha1_git is None or x.sha1_git in contents_to_add] | [x for x in content if x.sha1_git is None or x.sha1_git in contents_to_add] | ||||
) | ) | ||||
def directory_add(self, directories: List[Directory]) -> Dict: | def directory_add(self, directories: List[Directory]) -> Dict[str, int]: | ||||
missing_ids = self._filter_missing_ids("directory", (d.id for d in directories)) | missing_ids = self._filter_missing_ids("directory", (d.id for d in directories)) | ||||
return self.storage.directory_add( | return self.storage.directory_add( | ||||
[d for d in directories if d.id in missing_ids] | [d for d in directories if d.id in missing_ids] | ||||
) | ) | ||||
def revision_add(self, revisions: List[Revision]) -> Dict: | def revision_add(self, revisions: List[Revision]) -> Dict[str, int]: | ||||
missing_ids = self._filter_missing_ids("revision", (r.id for r in revisions)) | missing_ids = self._filter_missing_ids("revision", (r.id for r in revisions)) | ||||
return self.storage.revision_add([r for r in revisions if r.id in missing_ids]) | return self.storage.revision_add([r for r in revisions if r.id in missing_ids]) | ||||
def _filter_missing_contents(self, contents: List[Content]) -> Set[bytes]: | def _filter_missing_contents(self, contents: List[Content]) -> Set[bytes]: | ||||
"""Return only the content keys missing from swh | """Return only the content keys missing from swh | ||||
Args: | Args: | ||||
content_hashes: List of sha256 to check for existence in swh | content_hashes: List of sha256 to check for existence in swh | ||||
▲ Show 20 Lines • Show All 50 Lines • Show Last 20 Lines |