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: | ||||||||||||||||
ardumontUnsubmitted Done Inline Actions
ardumont: | ||||||||||||||||
cls: history's class, only 'prometheus' is supported actually | ||||||||||||||||
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) |