Changeset View
Changeset View
Standalone View
Standalone View
swh/counters/__init__.py
| # Copyright (C) 2021 The Software Heritage developers | # Copyright (C) 2021 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 | |||||||||||||||
| from __future__ import annotations | from __future__ import annotations | |||||||||||||||
| import importlib | import importlib | |||||||||||||||
| from typing import TYPE_CHECKING, Any, Dict | from typing import TYPE_CHECKING, Any, Dict | |||||||||||||||
| if TYPE_CHECKING: | if TYPE_CHECKING: | |||||||||||||||
| from swh.counters.interface import CountersInterface | from swh.counters.interface import CountersInterface, HistoryInterface | |||||||||||||||
| COUNTERS_IMPLEMENTATIONS = { | COUNTERS_IMPLEMENTATIONS = { | |||||||||||||||
| "redis": ".redis.Redis", | "redis": ".redis.Redis", | |||||||||||||||
| "remote": ".api.client.RemoteCounters", | "remote": ".api.client.RemoteCounters", | |||||||||||||||
| } | } | |||||||||||||||
| HISTORY_IMPLEMENTATIONS = { | ||||||||||||||||
| "prometheus": ".history.History", | ||||||||||||||||
| } | ||||||||||||||||
| def get_counters(cls: str, **kwargs: Dict[str, Any]) -> CountersInterface: | def get_counters(cls: str, **kwargs: Dict[str, Any]) -> CountersInterface: | |||||||||||||||
| """Get an counters object of class `cls` with arguments `args`. | """Get an counters object of class `cls` with arguments `args`. | |||||||||||||||
| Args: | Args: | |||||||||||||||
| cls: counters's class, either 'local' or 'remote' | cls: counters's class, either 'local' or 'remote' | |||||||||||||||
| args: dictionary of arguments passed to the | args: dictionary of arguments passed to the | |||||||||||||||
| counters class constructor | counters class constructor | |||||||||||||||
| Show All 10 Lines | if class_path is None: | |||||||||||||||
| "Unknown counters class `%s`. Supported: %s" | "Unknown counters class `%s`. Supported: %s" | |||||||||||||||
| % (cls, ", ".join(COUNTERS_IMPLEMENTATIONS)) | % (cls, ", ".join(COUNTERS_IMPLEMENTATIONS)) | |||||||||||||||
| ) | ) | |||||||||||||||
| (module_path, class_name) = class_path.rsplit(".", 1) | (module_path, class_name) = class_path.rsplit(".", 1) | |||||||||||||||
| module = importlib.import_module(module_path, package=__package__) | module = importlib.import_module(module_path, package=__package__) | |||||||||||||||
| Counters = getattr(module, class_name) | Counters = getattr(module, class_name) | |||||||||||||||
| return Counters(**kwargs) | return Counters(**kwargs) | |||||||||||||||
| def get_history(cls: str, **kwargs: Dict[str, Any]) -> HistoryInterface: | ||||||||||||||||
| """Get a history object of class `cls` with arguments `kwargs`. | ||||||||||||||||
| Args: | ||||||||||||||||
| cls: hystory's class, only 'prometheus' is supported actually | ||||||||||||||||
ardumontUnsubmitted Done Inline Actions
ardumont: | ||||||||||||||||
| kwargs: dictionary of arguments passed to the | ||||||||||||||||
| counters class constructor | ||||||||||||||||
Done Inline Actions
ardumont: | ||||||||||||||||
| Returns: | ||||||||||||||||
| an instance of swh.counters.history's classes (either local or remote) | ||||||||||||||||
| Raises: | ||||||||||||||||
| ValueError if passed an unknown history class. | ||||||||||||||||
| """ | ||||||||||||||||
| class_path = HISTORY_IMPLEMENTATIONS.get(cls) | ||||||||||||||||
| if class_path is None: | ||||||||||||||||
| raise ValueError( | ||||||||||||||||
| "Unknown history class `%s`. Supported: %s" | ||||||||||||||||
| % (cls, ", ".join(HISTORY_IMPLEMENTATIONS)) | ||||||||||||||||
| ) | ||||||||||||||||
| (module_path, class_name) = class_path.rsplit(".", 1) | ||||||||||||||||
| module = importlib.import_module(module_path, package=__package__) | ||||||||||||||||
| History = getattr(module, class_name) | ||||||||||||||||
| return History(**kwargs) | ||||||||||||||||