Page MenuHomeSoftware Heritage
Files F7437607

factory.py
No OneTemporary
Actions

factory.py
View Options

# Copyright (C) 2016-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
from typing import Callable, Dict, Union
from swh.objstorage.objstorage import ObjStorage, ID_HASH_LENGTH # noqa
from swh.objstorage.backends.pathslicing import PathSlicingObjStorage
from swh.objstorage.backends.in_memory import InMemoryObjStorage
from swh.objstorage.api.client import RemoteObjStorage
from swh.objstorage.multiplexer import MultiplexerObjStorage, StripingObjStorage
from swh.objstorage.multiplexer.filter import add_filters
from swh.objstorage.backends.seaweed import WeedObjStorage
from swh.objstorage.backends.generator import RandomGeneratorObjStorage
__all__ = ["get_objstorage", "ObjStorage"]
_STORAGE_CLASSES: Dict[str, Union[type, Callable[..., type]]] = {
"pathslicing": PathSlicingObjStorage,
"remote": RemoteObjStorage,
"memory": InMemoryObjStorage,
"weed": WeedObjStorage,
"random": RandomGeneratorObjStorage,
}
_STORAGE_CLASSES_MISSING = {}
try:
from swh.objstorage.backends.azure import (
AzureCloudObjStorage,
PrefixedAzureCloudObjStorage,
)
_STORAGE_CLASSES["azure"] = AzureCloudObjStorage
_STORAGE_CLASSES["azure-prefixed"] = PrefixedAzureCloudObjStorage
except ImportError as e:
_STORAGE_CLASSES_MISSING["azure"] = e.args[0]
_STORAGE_CLASSES_MISSING["azure-prefixed"] = e.args[0]
try:
from swh.objstorage.backends.rados import RADOSObjStorage
_STORAGE_CLASSES["rados"] = RADOSObjStorage
except ImportError as e:
_STORAGE_CLASSES_MISSING["rados"] = e.args[0]
try:
from swh.objstorage.backends.libcloud import (
AwsCloudObjStorage,
OpenStackCloudObjStorage,
)
_STORAGE_CLASSES["s3"] = AwsCloudObjStorage
_STORAGE_CLASSES["swift"] = OpenStackCloudObjStorage
except ImportError as e:
_STORAGE_CLASSES_MISSING["s3"] = e.args[0]
_STORAGE_CLASSES_MISSING["swift"] = e.args[0]
def get_objstorage(cls, args):
""" Create an ObjStorage using the given implementation class.
Args:
cls (str): objstorage class unique key contained in the
_STORAGE_CLASSES dict.
args (dict): arguments for the required class of objstorage
that must match exactly the one in the `__init__` method of the
class.
Returns:
subclass of ObjStorage that match the given `storage_class` argument.
Raises:
ValueError: if the given storage class is not a valid objstorage
key.
"""
if cls in _STORAGE_CLASSES:
return _STORAGE_CLASSES[cls](**args)
else:
raise ValueError(
"Storage class {} is not available: {}".format(
cls, _STORAGE_CLASSES_MISSING.get(cls, "unknown name")
)
)
def _construct_filtered_objstorage(storage_conf, filters_conf):
return add_filters(get_objstorage(**storage_conf), filters_conf)
_STORAGE_CLASSES["filtered"] = _construct_filtered_objstorage
def _construct_multiplexer_objstorage(objstorages):
storages = [get_objstorage(**conf) for conf in objstorages]
return MultiplexerObjStorage(storages)
_STORAGE_CLASSES["multiplexer"] = _construct_multiplexer_objstorage
def _construct_striping_objstorage(objstorages):
storages = [get_objstorage(**conf) for conf in objstorages]
return StripingObjStorage(storages)
_STORAGE_CLASSES["striping"] = _construct_striping_objstorage

File Metadata

Mime Type
text/x-python
Expires
Tue, Apr 15, 2:21 AM (2 d, 13 m ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3286220

Event Timeline

About · Developer information · Legal