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): # type: (...) -> 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): # type: (...) -> 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: | ): # type: (...) -> 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