0.0.168

orso/tools.py

@@ -18,6 +18,7 @@
 import threading
 import time
 import uuid
+from collections import OrderedDict
 from functools import wraps
 from random import getrandbits
 from typing import Any
@@ -438,22 +439,75 @@ def single_item_cache(
 
     @wraps(func)
     def wrapper(*args, **kwargs):
+        nonlocal cache
         current_time = time.time()
 
-        if (cache["last_args"] == args and cache["last_kwargs"] == kwargs) and (
-            current_time - cache["last_time"] <= valid_for_seconds
+        if (
+            cache["last_args"] == args
+            and cache["last_kwargs"] == kwargs
+            and current_time - cache["last_time"] <= valid_for_seconds
         ):
             return cache["last_result"]
 
         result = func(*args, **kwargs)
-        cache.update(
-            {
-                "last_args": args,  # type:ignore
-                "last_kwargs": kwargs,  # type:ignore
-                "last_result": result,
-                "last_time": current_time,  # type:ignore
-            }
+        cache["last_args"] = args
+        cache["last_kwargs"] = kwargs
+        cache["last_result"] = result
+        cache["last_time"] = current_time
+
+        return result
+
+    return wrapper
+
+
+def lru_cache_with_expiry(
+    func: Callable = None, *, max_size: int = 5, valid_for_seconds: float = float("inf")
+) -> Callable:
+    """
+    LRU cache decorator with optional expiration time and a fixed size.
+
+    Parameters:
+        func: Callable, optional
+            The function to be decorated.
+        maxsize: int, optional
+            The maximum size of the cache.
+        valid_for_seconds: float, optional
+            Number of seconds after which the cache expires.
+    """
+    if func is None:
+        return lambda f: lru_cache_with_expiry(
+            f, max_size=max_size, valid_for_seconds=valid_for_seconds
         )
+
+    cache: OrderedDict[Tuple[Any, ...], Tuple[float, Any]] = OrderedDict()
+
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        nonlocal cache
+        current_time = time.time()
+        key = (args, frozenset(kwargs.items()))
+
+        # Remove expired items
+        expired_keys = [
+            k for k, (timestamp, _) in cache.items() if current_time - timestamp > valid_for_seconds
+        ]
+        for k in expired_keys:
+            del cache[k]
+
+        # Check if result is cached
+        if key in cache:
+            # Move the accessed item to the end to maintain LRU order
+            cache.move_to_end(key)
+            return cache[key][1]
+
+        # Call the function and cache the result
+        result = func(*args, **kwargs)
+        cache[key] = (current_time, result)
+
+        # Maintain the cache size
+        if len(cache) > max_size:
+            cache.popitem(last=False)  # Remove the first (least recently used) item
+
         return result
 
     return wrapper

orso/version.py

@@ -10,5 +10,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version__: str = "0.0.167"
+__version__: str = "0.0.168"
 __author__: str = "@joocer"

tests/test_lru_with_expiration.py

@@ -0,0 +1,76 @@
+import os
+import sys
+
+
+sys.path.insert(1, os.path.join(sys.path[0], ".."))
+
+import pytest
+import time
+
+# Import the LRU cache decorator
+from orso.tools import lru_cache_with_expiry
+from orso.tools import random_string
+
+@lru_cache_with_expiry(max_size=3, valid_for_seconds=1)
+def sample_function(x, y):
+    return random_string()
+
+def test_cache_basic_functionality():
+    # Initial call, should not be cached
+    result1 = sample_function(1, 2)
+
+    # Call with same arguments, should be cached
+    result2 = sample_function(1, 2)
+    assert result2 == result1
+
+def test_cache_expiry():
+    # Initial call, should not be cached
+    result1 = sample_function(2, 3)
+
+    # Sleep for longer than the cache validity period
+    time.sleep(1)
+
+    # Call with same arguments, cache should have expired
+    result2 = sample_function(2, 3)
+    assert result2 != result1
+
+def test_cache_lru_eviction():
+    # Fill the cache with three different items
+    result1 = sample_function(1, 2)  # Should be cached
+    result2 = sample_function(2, 3)  # Should be cached
+    result3 = sample_function(3, 4)  # Should be cached
+
+    # At this point, the cache is full. The next item should cause the oldest item to be evicted
+    result4 = sample_function(4, 5)  # Should be cached, (1, 2) should be evicted
+
+    # Accessing the first item should result in a cache miss and recalculation
+    result1_again = sample_function(1, 2)
+
+    assert result1_again != result1  # New result as the old one should have been evicted
+
+    # Ensure the recently accessed items are still cached
+    result3_again = sample_function(3, 4)
+    assert result3_again == result3
+
+
+def test_cache_lru_access_order():
+    # Fill the cache with three different items
+    result1 = sample_function(5, 6)  # Should be cached
+    result2 = sample_function(6, 7)  # Should be cached
+    result3 = sample_function(7, 8)  # Should be cached
+
+    # Access the first item to make it recently used
+    result1_again = sample_function(5, 6)
+    assert result1_again == result1
+
+    # Add a new item to the cache, causing the LRU item to be evicted
+    result4 = sample_function(8, 9)  # Should be cached, (6, 7) should be evicted as it is now the LRU
+
+    # Accessing the evicted item should result in a cache miss and recalculation
+    result2_again = sample_function(6, 7)
+    assert result2_again != result2  # New result as the old one should have been evicted
+
+if __name__ == "__main__":  # prgama: nocover
+    from tests import run_tests
+
+    run_tests()

tests/test_profiler.py

@@ -102,11 +102,11 @@ def test_profile_estimators():
     missions = opteryx.query("SELECT * FROM $missions")
     profile: TableProfile = missions.profile
 
-    source = opteryx.query("SELECT COUNT(*) as missing FROM $missions WHERE Launched_at IS NULL")
+    source = opteryx.query("SELECT COUNT(*) as missing FROM $missions WHERE Lauched_at IS NULL")
     values = source.fetchone().as_dict
     assert (
-        profile.column("Launched_at").missing == values["missing"]
-    ), f"{profile.column('Launched_at').missing}, {values['missing']}"
+        profile.column("Lauched_at").missing == values["missing"]
+    ), f"{profile.column('Lauched_at').missing}, {values['missing']}"
 
     source = opteryx.query("SELECT MIN(Price) as minimum FROM $missions")
     values = source.fetchone().as_dict
@@ -127,9 +127,9 @@ def test_profile_estimators():
     assert profile.column("Company").most_frequent_values[0] == values["Company"], values
     assert profile.column("Company").most_frequent_counts[0] == values["frequency"], values
 
-    source = opteryx.query("SELECT COUNT_DISTINCT(Launched_at) AS unique_timestamps FROM $missions")
+    source = opteryx.query("SELECT COUNT_DISTINCT(Lauched_at) AS unique_timestamps FROM $missions")
     values = source.fetchone().as_dict
-    estimated_cardinality = profile.column("Launched_at").estimate_cardinality()
+    estimated_cardinality = profile.column("Lauched_at").estimate_cardinality()
     assert (
         estimated_cardinality * 0.75 < values["unique_timestamps"] < estimated_cardinality * 1.25
     ), f"{profile.column('Launched_at').estimate_cardinality()} != {values['unique_timestamps']}"