Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/__init__.py
# Copyright (C) 2015-2020 The Software Heritage developers | # Copyright (C) 2015-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 General Public License version 3, or any later version | # License: GNU 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 | ||||
import importlib | import importlib | ||||
from typing import Any, Dict, List | from typing import Any, Dict, List, TYPE_CHECKING | ||||
import warnings | import warnings | ||||
if TYPE_CHECKING: | |||||
from .interface import StorageInterface | from .interface import StorageInterface | ||||
STORAGE_IMPLEMENTATIONS = { | STORAGE_IMPLEMENTATIONS = { | ||||
"local": ".postgresql.storage.Storage", | "local": ".postgresql.storage.Storage", | ||||
"remote": ".api.client.RemoteStorage", | "remote": ".api.client.RemoteStorage", | ||||
"memory": ".in_memory.InMemoryStorage", | "memory": ".in_memory.InMemoryStorage", | ||||
"filter": ".filter.FilteringProxyStorage", | "filter": ".filter.FilteringProxyStorage", | ||||
"buffer": ".buffer.BufferingProxyStorage", | "buffer": ".buffer.BufferingProxyStorage", | ||||
"retry": ".retry.RetryingProxyStorage", | "retry": ".retry.RetryingProxyStorage", | ||||
"cassandra": ".cassandra.CassandraStorage", | "cassandra": ".cassandra.CassandraStorage", | ||||
"validate": ".validate.ValidatingProxyStorage", | "validate": ".validate.ValidatingProxyStorage", | ||||
} | } | ||||
def get_storage(cls: str, **kwargs) -> StorageInterface: | def get_storage(cls: str, **kwargs) -> "StorageInterface": | ||||
vlorentz: string | |||||
"""Get a storage object of class `storage_class` with arguments | """Get a storage object of class `storage_class` with arguments | ||||
`storage_args`. | `storage_args`. | ||||
Args: | Args: | ||||
storage (dict): dictionary with keys: | storage (dict): dictionary with keys: | ||||
- cls (str): storage's class, either local, remote, memory, filter, | - cls (str): storage's class, either local, remote, memory, filter, | ||||
buffer | buffer | ||||
- args (dict): dictionary with keys | - args (dict): dictionary with keys | ||||
Show All 30 Lines | def get_storage(cls: str, **kwargs) -> "StorageInterface": | ||||
if check_config: | if check_config: | ||||
if not storage.check_config(**check_config): | if not storage.check_config(**check_config): | ||||
raise EnvironmentError("storage check config failed") | raise EnvironmentError("storage check config failed") | ||||
return storage | return storage | ||||
def get_storage_pipeline( | def get_storage_pipeline( | ||||
steps: List[Dict[str, Any]], check_config=None | steps: List[Dict[str, Any]], check_config=None | ||||
) -> StorageInterface: | ) -> "StorageInterface": | ||||
Not Done Inline Actionsstring vlorentz: string | |||||
"""Recursively get a storage object that may use other storage objects | """Recursively get a storage object that may use other storage objects | ||||
as backends. | as backends. | ||||
Args: | Args: | ||||
steps (List[dict]): List of dicts that may be used as kwargs for | steps (List[dict]): List of dicts that may be used as kwargs for | ||||
`get_storage`. | `get_storage`. | ||||
Returns: | Returns: | ||||
Show All 25 Lines |
string