swh.objstorage.factory module#

swh.objstorage.factory.get_objstorage(cls: str, **kwargs) ObjStorageInterface[source]#

Create an ObjStorage using the given implementation class.

  • cls – objstorage class unique key contained in the OBJSTORAGE_IMPLEMENTATIONS dict.

  • kwargs – arguments for the required class of objstorage that must match exactly the one in the __init__ method of the class.


subclass of ObjStorage that match the given storage_class argument.


ValueError – if the given storage class is not a valid objstorage key.

class swh.objstorage.factory.ObjStorage(*, allow_delete: bool = False, **kwargs)[source]#

Bases: object

PRIMARY_HASH: Literal['sha1', 'sha256'] = 'sha1'#
compression: Literal['bz2', 'lzma', 'gzip', 'zlib', 'none'] = 'none'#
name: str = 'objstorage'#

Default objstorage name; can be overloaded at instantiation time giving a ‘name’ argument to the constructor

add_batch(contents: Mapping[bytes, bytes] | Iterable[Tuple[bytes | CompositeObjId, bytes]], check_presence: bool = True) Dict[source]#
restore(content: bytes, obj_id: bytes | CompositeObjId) None[source]#
get_batch(obj_ids: Iterable[bytes | CompositeObjId]) Iterator[bytes | None][source]#
abstract delete(obj_id: bytes | CompositeObjId)[source]#
list_content(last_obj_id: bytes | CompositeObjId | None = None, limit: int | None = 10000) Iterator[CompositeObjId][source]#
download_url(obj_id: bytes | CompositeObjId, content_disposition: str | None = None, expiry: timedelta | None = None) str | None[source]#
abstract get(obj_id: bytes | CompositeObjId) bytes[source]#
check(obj_id: bytes | CompositeObjId) None[source]#

Check if a content is found and recompute its hash to check integrity.

compress(data: bytes) bytes[source]#
decompress(data: bytes, hex_obj_id: str) bytes[source]#