Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion src/usepy/decorator/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
]
71 changes: 71 additions & 0 deletions src/usepy/decorator/debounce.py
Original file line number Diff line number Diff line change
@@ -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)
57 changes: 57 additions & 0 deletions src/usepy/decorator/memoize.py
Original file line number Diff line number Diff line change
@@ -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
139 changes: 139 additions & 0 deletions tests/test_decorator_new.py
Original file line number Diff line number Diff line change
@@ -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
Loading