feat(utils): add convert_size

replete/utils.py

@@ -3,6 +3,7 @@
 import functools
 import itertools
 import logging
+import math
 import weakref
 from concurrent import futures
 from typing import TYPE_CHECKING, Any, Callable, Hashable, Iterable, Iterator, Mapping, Sequence, TypeVar
@@ -174,6 +175,7 @@ def func_with_idx(idx, *args, **kwargs) -> tuple[int, Any]:
                 yield func_result
 
 
+# TODO: This function doesn't seem to work with multiple instances of a class, need to debug
 def weak_lru_cache(maxsize: Callable | int | None = 128, *, typed=False):
     """
     LRU Cache decorator that keeps a weak reference to 'self'
@@ -199,3 +201,13 @@ def inner(self, *args, **kwargs):
         raise ValueError("Expected maxsize to be an integer or None")
 
     return functools.partial(helper, maxsize, typed=typed)
+
+
+def convert_size(size_bytes):
+    if size_bytes == 0:
+        return "0B"
+    size_name = ("B", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB")
+    i = int(math.floor(math.log(size_bytes, 1024)))
+    p = math.pow(1024, i)
+    s = round(size_bytes / p, 2)
+    return f"{s} {size_name[i]}"

tests/test_utils.py

@@ -9,7 +9,7 @@
 from flaky import flaky
 
 from replete import Timer, ensure_unique_keys, split_list
-from replete.utils import futures_processing, weak_lru_cache
+from replete.utils import convert_size, futures_processing, weak_lru_cache
 
 
 def test_unique_keys():
@@ -127,3 +127,16 @@ def func() -> None:
     gc.collect()  # collect garbage
     # Since foo went out of scope after func() finished, it should be garbage collected
     assert len([obj for obj in gc.get_objects() if isinstance(obj, WeakCacheTester)]) == 0
+
+
+@pytest.mark.parametrize(
+    ("size", "string"),
+    [
+        (2, "2.0 B"),
+        (1024, "1.0 KiB"),
+        (3 * 1024**2, "3.0 MiB"),
+        (4.31 * 1024**3, "4.31 GiB"),
+    ],
+)
+def test_convert_size(size: int, string: str):
+    assert convert_size(size) == string