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 Any, 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 default_cfg = { "default_interval": "1 day", "min_interval": "1 day", "max_interval": "1 day", "backoff_factor": 1, "max_queue_length": 10000, }
[docs] def register_tasks() -> Dict[str, Any]: return { "task_modules": [f"{__name__}.cooking_tasks"], "task_types": { "vault-cook-bundle": default_cfg, "vault-batch-cook-bundle": default_cfg, }, }