Page MenuHomeSoftware Heritage
Files F9336843

No OneTemporary
Actions

View Options

diff --git a/swh/counters/__init__.py b/swh/counters/__init__.py
index bcc234d..d762c6f 100644
--- a/swh/counters/__init__.py
+++ b/swh/counters/__init__.py
@@ -1,75 +1,76 @@
# Copyright (C) 2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
from __future__ import annotations
import importlib
from typing import TYPE_CHECKING, Any, Dict
if TYPE_CHECKING:
from swh.counters.interface import CountersInterface, HistoryInterface
COUNTERS_IMPLEMENTATIONS = {
"redis": ".redis.Redis",
"remote": ".api.client.RemoteCounters",
+ "memory": ".in_memory.InMemory",
}
HISTORY_IMPLEMENTATIONS = {
"prometheus": ".history.History",
}
def get_counters(cls: str, **kwargs: Dict[str, Any]) -> CountersInterface:
"""Get an counters object of class `cls` with arguments `args`.
Args:
cls: counters's class, either 'local' or 'remote'
args: dictionary of arguments passed to the
counters class constructor
Returns:
an instance of swh.counters's classes (either local or remote)
Raises:
ValueError if passed an unknown counters class.
"""
class_path = COUNTERS_IMPLEMENTATIONS.get(cls)
if class_path is None:
raise ValueError(
"Unknown counters class `%s`. Supported: %s"
% (cls, ", ".join(COUNTERS_IMPLEMENTATIONS))
)
(module_path, class_name) = class_path.rsplit(".", 1)
module = importlib.import_module(module_path, package=__package__)
Counters = getattr(module, class_name)
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: history's class, only 'prometheus' is supported actually
kwargs: dictionary of arguments passed to the
counters class constructor
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)
diff --git a/swh/counters/in_memory.py b/swh/counters/in_memory.py
new file mode 100644
index 0000000..d4c4889
--- /dev/null
+++ b/swh/counters/in_memory.py
@@ -0,0 +1,31 @@
+# Copyright (C) 2021 The Software Heritage developers
+# See the AUTHORS file at the top-level directory of this distribution
+# License: GNU General Public License version 3, or any later version
+# See top-level LICENSE file for more information
+
+from collections import defaultdict
+from typing import Any, Dict, Iterable, List
+
+
+class InMemory:
+ """InMemory implementation of the counters.
+ Naive implementation using a Dict[str, Set]"""
+
+ def __init__(self):
+ self.counters = defaultdict(set)
+
+ def check(self):
+ return "OK"
+
+ def add(self, collection: str, keys: Iterable[Any]) -> None:
+ for value in keys:
+ self.counters[collection].add(value)
+
+ def get_count(self, collection: str) -> int:
+ return len(self.counters.get(collection, []))
+
+ def get_counts(self, collections: List[str]) -> Dict[str, int]:
+ return {coll: self.get_count(coll) for coll in collections}
+
+ def get_counters(self) -> Iterable[str]:
+ return list(self.counters.keys())
diff --git a/swh/counters/tests/test_init.py b/swh/counters/tests/test_init.py
index 18d38d5..1eb5915 100644
--- a/swh/counters/tests/test_init.py
+++ b/swh/counters/tests/test_init.py
@@ -1,90 +1,92 @@
# Copyright (C) 2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
import inspect
import pytest
from swh.counters import get_counters, get_history
from swh.counters.api.client import RemoteCounters
from swh.counters.history import History
+from swh.counters.in_memory import InMemory
from swh.counters.interface import CountersInterface
from swh.counters.redis import Redis
COUNTERS_IMPLEMENTATIONS = [
("remote", RemoteCounters, {"url": "localhost"}),
("redis", Redis, {"host": "localhost"}),
+ ("memory", InMemory, {}),
]
def test_get_counters_failure():
with pytest.raises(ValueError, match="Unknown counters class"):
get_counters("unknown-counters")
@pytest.mark.parametrize("class_,expected_class,kwargs", COUNTERS_IMPLEMENTATIONS)
def test_get_counters(mocker, class_, expected_class, kwargs):
if kwargs:
concrete_counters = get_counters(class_, **kwargs)
else:
concrete_counters = get_counters(class_)
assert isinstance(concrete_counters, expected_class)
@pytest.mark.parametrize("class_,expected_class,kwargs", COUNTERS_IMPLEMENTATIONS)
def test_types(mocker, class_, expected_class, kwargs):
"""Checks all methods of CountersInterface are implemented by this
backend, and that they have the same signature.
"""
# mocker.patch("swh.counters.redis.Redis")
if kwargs:
concrete_counters = get_counters(class_, **kwargs)
else:
concrete_counters = get_counters(class_)
# Create an instance of the protocol (which cannot be instantiated
# directly, so this creates a subclass, then instantiates it)
interface = type("_", (CountersInterface,), {})()
for meth_name in dir(interface):
if meth_name.startswith("_"):
continue
interface_meth = getattr(interface, meth_name)
missing_methods = []
try:
concrete_meth = getattr(concrete_counters, meth_name)
except AttributeError:
if not getattr(interface_meth, "deprecated_endpoint", False):
# The backend is missing a (non-deprecated) endpoint
missing_methods.append(meth_name)
continue
expected_signature = inspect.signature(interface_meth)
actual_signature = inspect.signature(concrete_meth)
assert expected_signature == actual_signature, meth_name
assert missing_methods == []
def test_get_history_failure():
with pytest.raises(ValueError, match="Unknown history class"):
get_history("unknown-history")
def test_get_history():
concrete_history = get_history(
"prometheus",
**{
"prometheus_host": "",
"prometheus_port": "",
"live_data_start": "",
"cache_base_directory": "",
},
)
assert isinstance(concrete_history, History)
diff --git a/swh/counters/tests/test_inmemory.py b/swh/counters/tests/test_inmemory.py
new file mode 100644
index 0000000..1380350
--- /dev/null
+++ b/swh/counters/tests/test_inmemory.py
@@ -0,0 +1,54 @@
+# Copyright (C) 2021 The Software Heritage developers
+# See the AUTHORS file at the top-level directory of this distribution
+# License: GNU General Public License version 3, or any later version
+# See top-level LICENSE file for more information
+
+from swh.counters.in_memory import InMemory
+
+
+def test__inmemory__add():
+ im = InMemory()
+
+ im.add("counter1", ["val1"])
+ im.add("counter2", ["val1"])
+ im.add("counter1", [1])
+
+ assert im.get_count("counter1") == 2
+ assert im.get_count("counter2") == 1
+ assert im.get_count("inexisting") == 0
+
+
+def test__inmemory_getcounters():
+ im = InMemory()
+
+ assert len(im.get_counters()) == 0
+
+ counters = ["c1", "c2", "c3"]
+
+ count = 0
+
+ for counter in counters:
+ im.add(counter, [1, 2])
+ count += 1
+ assert count == len(im.get_counters())
+
+ results = im.get_counters()
+ assert results == counters
+
+
+def test__inmemory_getcounts():
+ im = InMemory()
+
+ expected = {"c1": 1, "c2": 2, "c3": 0}
+
+ im.add("c1", ["v1"])
+ im.add("c2", ["v1", "v2"])
+
+ result = im.get_counts(["c1", "c2", "c3"])
+ assert result == expected
+
+
+def test__inmemory_check():
+ im = InMemory()
+
+ assert im.check() == "OK"

File Metadata

Mime Type
text/x-diff
Expires
Jul 4 2025, 7:46 AM (10 w, 4 d ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3236140

Event Timeline

About · Developer information · Legal