From 0eff3c0103bacc39a87fcde84fe055863790dc48 Mon Sep 17 00:00:00 2001 From: mic1on Date: Sat, 14 Mar 2026 13:36:20 +0800 Subject: [PATCH] feat(decorator): add 2 new decorators - debounce: delay execution until after specified time - memoize: cache function results by arguments With comprehensive unit tests (7 test cases) --- src/usepy/decorator/__init__.py | 13 ++- src/usepy/decorator/debounce.py | 71 ++++++++++++++++ src/usepy/decorator/memoize.py | 57 +++++++++++++ tests/test_decorator_new.py | 139 ++++++++++++++++++++++++++++++++ 4 files changed, 279 insertions(+), 1 deletion(-) create mode 100644 src/usepy/decorator/debounce.py create mode 100644 src/usepy/decorator/memoize.py create mode 100644 tests/test_decorator_new.py diff --git a/src/usepy/decorator/__init__.py b/src/usepy/decorator/__init__.py index 7376ffb..aaf46d1 100644 --- a/src/usepy/decorator/__init__.py +++ b/src/usepy/decorator/__init__.py @@ -2,5 +2,16 @@ from .catch_error import catch_error from .singleton import singleton from .throttle import Throttle as throttle +from .debounce import debounce, Debounce +from .memoize import memoize -__all__ = ["retry", "catch_error", "singleton", "throttle"] +__all__ = [ + "retry", + "catch_error", + "singleton", + "throttle", + # New decorators + "debounce", + "Debounce", + "memoize", +] \ No newline at end of file diff --git a/src/usepy/decorator/debounce.py b/src/usepy/decorator/debounce.py new file mode 100644 index 0000000..be46824 --- /dev/null +++ b/src/usepy/decorator/debounce.py @@ -0,0 +1,71 @@ +import asyncio +from functools import wraps +import time +import threading + + +class Debounce: + """ + Debounce Decorator + + Delays the function execution until after a specified time has elapsed + since the last call. Useful for rate-limiting rapid function calls. + + Args: + delay (float): The delay in seconds to wait before executing. + + Example: + >>> @debounce(delay=0.5) + ... def search(query): + ... print(f"Searching for: {query}") + """ + + def __init__(self, delay: float): + self.delay = delay + self.timer = None + self.lock = threading.Lock() + + def __call__(self, func): + @wraps(func) + def wrapper(*args, **kwargs): + with self.lock: + if self.timer: + self.timer.cancel() + + def execute(): + return func(*args, **kwargs) + + self.timer = threading.Timer(self.delay, execute) + self.timer.start() + + @wraps(func) + async def async_wrapper(*args, **kwargs): + with self.lock: + if self.timer: + self.timer.cancel() + + async def execute(): + return await func(*args, **kwargs) + + self.timer = threading.Timer(self.delay, lambda: asyncio.run(execute())) + self.timer.start() + + return async_wrapper if asyncio.iscoroutinefunction(func) else wrapper + + +def debounce(delay: float): + """ + Debounce decorator factory. + + Args: + delay (float): The delay in seconds to wait before executing. + + Returns: + Debounce: The debounce decorator. + + Example: + >>> @debounce(delay=0.5) + ... def save_data(data): + ... print(f"Saving: {data}") + """ + return Debounce(delay) \ No newline at end of file diff --git a/src/usepy/decorator/memoize.py b/src/usepy/decorator/memoize.py new file mode 100644 index 0000000..7acd170 --- /dev/null +++ b/src/usepy/decorator/memoize.py @@ -0,0 +1,57 @@ +import asyncio +from functools import wraps +from typing import Any, Callable, Dict, Tuple +import hashlib +import pickle + + +def memoize(func: Callable) -> Callable: + """ + Memoize Decorator + + Caches the results of function calls based on arguments. + Subsequent calls with the same arguments return the cached result. + + Args: + func (Callable): The function to memoize. + + Returns: + Callable: The memoized function. + + Example: + >>> @memoize + ... def expensive_computation(n): + ... print(f"Computing {n}...") + ... return n * n + >>> expensive_computation(5) + Computing 5... + 25 + >>> expensive_computation(5) # Returns cached result + 25 + """ + cache: Dict[str, Any] = {} + + def _make_key(args: Tuple, kwargs: Dict) -> str: + """Generate a unique key for the given arguments.""" + key_data = pickle.dumps((args, frozenset(kwargs.items()))) + return hashlib.md5(key_data).hexdigest() + + @wraps(func) + def wrapper(*args, **kwargs): + key = _make_key(args, kwargs) + if key not in cache: + cache[key] = func(*args, **kwargs) + return cache[key] + + @wraps(func) + async def async_wrapper(*args, **kwargs): + key = _make_key(args, kwargs) + if key not in cache: + cache[key] = await func(*args, **kwargs) + return cache[key] + + # Add method to clear cache + wrapper.cache_clear = lambda: cache.clear() + async_wrapper.cache_clear = lambda: cache.clear() + + return async_wrapper if asyncio.iscoroutinefunction(func) else wrapper \ No newline at end of file diff --git a/tests/test_decorator_new.py b/tests/test_decorator_new.py new file mode 100644 index 0000000..e4db1e6 --- /dev/null +++ b/tests/test_decorator_new.py @@ -0,0 +1,139 @@ +import pytest +import time +import asyncio +from usepy.decorator import debounce, memoize + + +class TestDebounce: + """Tests for debounce decorator""" + + def test_debounce_delays_execution(self): + """Test that debounce delays execution""" + results = [] + + @debounce(delay=0.1) + def add_result(value): + results.append(value) + + add_result(1) + add_result(2) + add_result(3) + + # Immediately, no results yet + assert results == [] + + # Wait for debounce + time.sleep(0.15) + assert results == [3] # Only last call executed + + def test_debounce_multiple_calls(self): + """Test multiple debounced calls""" + results = [] + + @debounce(delay=0.1) + def record(value): + results.append(value) + + record('a') + time.sleep(0.15) + assert results == ['a'] + + record('b') + time.sleep(0.15) + assert results == ['a', 'b'] + + +class TestMemoize: + """Tests for memoize decorator""" + + def test_memoize_caches_result(self): + """Test that memoize caches results""" + call_count = 0 + + @memoize + def expensive(n): + nonlocal call_count + call_count += 1 + return n * n + + # First call computes + result1 = expensive(5) + assert result1 == 25 + assert call_count == 1 + + # Second call uses cache + result2 = expensive(5) + assert result2 == 25 + assert call_count == 1 # Not incremented + + def test_memoize_different_args(self): + """Test memoize with different arguments""" + call_count = 0 + + @memoize + def compute(n): + nonlocal call_count + call_count += 1 + return n * 2 + + compute(1) + compute(2) + compute(1) # Cached + + assert call_count == 2 # Only 2 computations + + def test_memoize_with_kwargs(self): + """Test memoize with keyword arguments""" + call_count = 0 + + @memoize + def greet(name, greeting='Hello'): + nonlocal call_count + call_count += 1 + return f"{greeting}, {name}!" + + result1 = greet('World') + result2 = greet('World') + result3 = greet('World', greeting='Hi') + + assert result1 == "Hello, World!" + assert result2 == "Hello, World!" + assert result3 == "Hi, World!" + assert call_count == 2 # Two different argument sets + + def test_memoize_clear_cache(self): + """Test clearing memoize cache""" + call_count = 0 + + @memoize + def compute(n): + nonlocal call_count + call_count += 1 + return n + + compute(1) + compute(1) + assert call_count == 1 + + # Clear cache + compute.cache_clear() + + compute(1) + assert call_count == 2 # Recomputed after clear + + def test_memoize_async(self): + """Test memoize with async function""" + call_count = 0 + + @memoize + async def async_compute(n): + nonlocal call_count + call_count += 1 + return n * n + + result1 = asyncio.run(async_compute(5)) + result2 = asyncio.run(async_compute(5)) + + assert result1 == 25 + assert result2 == 25 + assert call_count == 1 \ No newline at end of file