Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/multiplexer/filter/id_filter.py
Show All 29 Lines | class IdObjStorageFilter(ObjStorageFilter): | ||||
def is_valid(self, obj_id): | def is_valid(self, obj_id): | ||||
""" Indicates if the given id is valid. | """ Indicates if the given id is valid. | ||||
""" | """ | ||||
raise NotImplementedError('Implementations of an IdObjStorageFilter ' | raise NotImplementedError('Implementations of an IdObjStorageFilter ' | ||||
'must have a "is_valid" method') | 'must have a "is_valid" method') | ||||
def __contains__(self, obj_id, *args, **kwargs): | def __contains__(self, obj_id, *args, **kwargs): | ||||
""" Indicates if the given object is present in the storage | |||||
See base class [ObjStorage]. | |||||
""" | |||||
if self.is_valid(obj_id): | if self.is_valid(obj_id): | ||||
return self.storage.__contains__(*args, obj_id=obj_id, **kwargs) | return self.storage.__contains__(*args, obj_id=obj_id, **kwargs) | ||||
return False | return False | ||||
def __len__(self): | def __len__(self): | ||||
""" See base class [ObjStorageFilter]. | |||||
""" | |||||
return sum(1 for i in [id for id in self.storage if self.is_valid(id)]) | return sum(1 for i in [id for id in self.storage if self.is_valid(id)]) | ||||
def __iter__(self): | def __iter__(self): | ||||
""" See base class [ObjStorageFilter]. | |||||
""" | |||||
yield from filter(lambda id: self.is_valid(id), iter(self.storage)) | yield from filter(lambda id: self.is_valid(id), iter(self.storage)) | ||||
def add(self, content, obj_id=None, check_presence=True, *args, **kwargs): | def add(self, content, obj_id=None, check_presence=True, *args, **kwargs): | ||||
""" See base class [ObjStorageFilter]. | |||||
""" | |||||
if obj_id is None: | if obj_id is None: | ||||
obj_id = compute_hash(content) | obj_id = compute_hash(content) | ||||
if self.is_valid(obj_id): | if self.is_valid(obj_id): | ||||
return self.storage.add(content, *args, obj_id=obj_id, **kwargs) | return self.storage.add(content, *args, obj_id=obj_id, **kwargs) | ||||
def restore(self, content, obj_id=None, *args, **kwargs): | def restore(self, content, obj_id=None, *args, **kwargs): | ||||
""" See base class [ObjStorageFilter]. | |||||
""" | |||||
if obj_id is None: | if obj_id is None: | ||||
obj_id = compute_hash(content) | obj_id = compute_hash(content) | ||||
if self.is_valid(obj_id): | if self.is_valid(obj_id): | ||||
return self.storage.restore(content, *args, | return self.storage.restore(content, *args, | ||||
obj_id=obj_id, **kwargs) | obj_id=obj_id, **kwargs) | ||||
def get(self, obj_id, *args, **kwargs): | def get(self, obj_id, *args, **kwargs): | ||||
""" See base class [ObjStorageFilter]. | |||||
""" | |||||
if self.is_valid(obj_id): | if self.is_valid(obj_id): | ||||
return self.storage.get(*args, obj_id=obj_id, **kwargs) | return self.storage.get(*args, obj_id=obj_id, **kwargs) | ||||
raise ObjNotFoundError(obj_id) | raise ObjNotFoundError(obj_id) | ||||
def check(self, obj_id, *args, **kwargs): | def check(self, obj_id, *args, **kwargs): | ||||
""" See base class [ObjStorageFilter]. | |||||
""" | |||||
if self.is_valid(obj_id): | if self.is_valid(obj_id): | ||||
return self.storage.check(*args, obj_id=obj_id, **kwargs) | return self.storage.check(*args, obj_id=obj_id, **kwargs) | ||||
raise ObjNotFoundError(obj_id) | raise ObjNotFoundError(obj_id) | ||||
def get_random(self, *args, **kwargs): | def get_random(self, *args, **kwargs): | ||||
""" See base class [ObjStorageFilter]. | |||||
""" | |||||
yield from filter(lambda id: self.is_valid(id), | yield from filter(lambda id: self.is_valid(id), | ||||
self.storage.get_random(*args, **kwargs)) | self.storage.get_random(*args, **kwargs)) | ||||
class RegexIdObjStorageFilter(IdObjStorageFilter): | class RegexIdObjStorageFilter(IdObjStorageFilter): | ||||
""" Filter that allow operations if the content's id as hex match a regex. | """ Filter that allow operations if the content's id as hex match a regex. | ||||
""" | """ | ||||
Show All 20 Lines |