Changeset View
Changeset View
Standalone View
Standalone View
swh/vault/__init__.py
# Copyright (C) 2018 The Software Heritage developers | # Copyright (C) 2018-2020 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU Affero General Public License version 3, or any later version | # License: GNU Affero General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
from __future__ import annotations | |||||
import importlib | |||||
import logging | import logging | ||||
from typing import Dict | |||||
import warnings | |||||
logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
def get_vault(cls="remote", args={}): | BACKEND_TYPES: Dict[str, str] = { | ||||
"remote": ".api.client.RemoteVaultClient", | |||||
"local": ".backend.VaultBackend", | |||||
} | |||||
def get_vault(cls: str = "remote", **kwargs): | |||||
""" | """ | ||||
Get a vault object of class `vault_class` with arguments | Get a vault object of class `vault_class` with arguments | ||||
`vault_args`. | `vault_args`. | ||||
Args: | Args: | ||||
vault (dict): dictionary with keys: | cls: vault's class, either 'remote' or 'local' | ||||
- cls (str): vault's class, either 'remote' | kwargs: arguments to pass to the class' constructor | ||||
- args (dict): dictionary with keys | |||||
Returns: | Returns: | ||||
an instance of VaultBackend (either local or remote) | an instance of VaultBackend (either local or remote) | ||||
Raises: | Raises: | ||||
ValueError if passed an unknown storage class. | ValueError if passed an unknown storage class. | ||||
""" | """ | ||||
if cls == "remote": | if "args" in kwargs: | ||||
from .api.client import RemoteVaultClient as Vault | warnings.warn( | ||||
elif cls == "local": | 'Explicit "args" key is deprecated, use keys directly instead.', | ||||
from swh.scheduler import get_scheduler | DeprecationWarning, | ||||
from swh.storage import get_storage | ) | ||||
from swh.vault.backend import VaultBackend as Vault | kwargs = kwargs["args"] | ||||
from swh.vault.cache import VaultCache | |||||
class_path = BACKEND_TYPES.get(cls) | |||||
args["cache"] = VaultCache(**args["cache"]) | if class_path is None: | ||||
args["storage"] = get_storage(**args["storage"]) | raise ValueError( | ||||
args["scheduler"] = get_scheduler(**args["scheduler"]) | f"Unknown Vault class `{cls}`. " f"Supported: {', '.join(BACKEND_TYPES)}" | ||||
else: | ) | ||||
raise ValueError("Unknown storage class `%s`" % cls) | |||||
logger.debug("Instantiating %s with %s" % (Vault, args)) | (module_path, class_name) = class_path.rsplit(".", 1) | ||||
return Vault(**args) | module = importlib.import_module(module_path, package=__package__) | ||||
Vault = getattr(module, class_name) | |||||
return Vault(**kwargs) |