Source code for swh.vault

# Copyright (C) 2018-2022  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information

from __future__ import annotations

import importlib
import logging
from typing import Dict
import warnings

from swh.vault.backend import VaultDB

logger = logging.getLogger(__name__)

BACKEND_TYPES: Dict[str, str] = {
    "remote": ".api.client.RemoteVaultClient",
    "postgresql": ".backend.VaultBackend",
    "memory": ".in_memory_backend.InMemoryVaultBackend",
    # deprecated
    "local": ".backend.VaultBackend",

[docs]def get_vault(cls: str = "remote", **kwargs): """ Get a vault object of class `vault_class` with arguments `vault_args`. Args: cls: vault's class, either 'remote' or 'local' kwargs: arguments to pass to the class' constructor Returns: an instance of VaultBackend (either local or remote) Raises: ValueError if passed an unknown storage class. """ if "args" in kwargs: warnings.warn( 'Explicit "args" key is deprecated, use keys directly instead.', DeprecationWarning, ) kwargs = kwargs["args"] class_path = BACKEND_TYPES.get(cls) if class_path is None: raise ValueError( f"Unknown Vault class `{cls}`. " f"Supported: {', '.join(BACKEND_TYPES)}" ) (module_path, class_name) = class_path.rsplit(".", 1) module = importlib.import_module(module_path, package=__package__) Vault = getattr(module, class_name) return Vault(**kwargs)
get_datastore = VaultDB