Source code for swh.objstorage.factory

# Copyright (C) 2016-2024  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

import importlib
import warnings

from swh.objstorage.interface import ObjStorageInterface
from swh.objstorage.objstorage import ObjStorage

__all__ = ["get_objstorage", "ObjStorage"]

    "pathslicing": "swh.objstorage.backends.pathslicing.PathSlicingObjStorage",
    "remote": "swh.objstorage.api.client.RemoteObjStorage",
    "memory": "swh.objstorage.backends.in_memory.InMemoryObjStorage",
    "seaweedfs": "swh.objstorage.backends.seaweedfs.objstorage.SeaweedFilerObjStorage",
    "random": "swh.objstorage.backends.generator.RandomGeneratorObjStorage",
    "http": "swh.objstorage.backends.http.HTTPReadOnlyObjStorage",
    "noop": "swh.objstorage.backends.noop.NoopObjStorage",
    "azure": "",
    "azure-prefixed": "",
    "s3": "swh.objstorage.backends.libcloud.AwsCloudObjStorage",
    "swift": "swh.objstorage.backends.libcloud.OpenStackCloudObjStorage",
    "winery": "swh.objstorage.backends.winery.WineryObjStorage",
    # filters and proxies
    "multiplexer": "swh.objstorage.multiplexer.MultiplexerObjStorage",
    "read-only": "swh.objstorage.proxies.readonly.ReadOnlyProxyObjStorage",
    # deprecated factories
    "filtered": "_construct_filtered_objstorage",

[docs] def get_objstorage(cls: str, **kwargs) -> ObjStorageInterface: """Create an ObjStorage using the given implementation class. Args: 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. Returns: subclass of ObjStorage that match the given `storage_class` argument. Raises: ValueError: if the given storage class is not a valid objstorage key. """ class_path = OBJSTORAGE_IMPLEMENTATIONS.get(cls) if class_path is None: raise ValueError( "Unknown storage class `%s`. Supported: %s" % (cls, ", ".join(OBJSTORAGE_IMPLEMENTATIONS)) ) if "." in class_path: (module_path, class_name) = class_path.rsplit(".", 1) try: module = importlib.import_module(module_path, package=__package__) except ImportError as e: raise ValueError(f"Storage class {cls} is not available: {e.args[0]}") ObjStorage = getattr(module, class_name) else: # used by (deprecated) filtered for which the value is a factory # function rather than a class ObjStorage = globals()[class_path] return ObjStorage(**kwargs)
def _construct_filtered_objstorage(storage_conf, filters_conf, **kwargs): if len(filters_conf) != 1 or filters_conf[0]["type"] != "readonly": raise ValueError("This legacy function only supports a single readonly filter") warnings.warn( "The 'filtered[type:readonly]' objstorage class has been deprecated, " "please use a 'read-only' proxy class instead.", DeprecationWarning, ) return get_objstorage(cls="read-only", storage=get_objstorage(**storage_conf))