44
55
66class CachedFunction [** P , R ]:
7- """Wrapper class for a cached function with a clear_cache method."""
7+ """Wrapper class for a cached function with a clear_cache method.
88
9- def __init__ (self , func : Callable [P , R ], cache : Any ) -> None :
9+ This class wraps a function to provide TTL-based caching. The cache is shared
10+ across all instances when used as an instance method decorator.
11+
12+ Example:
13+ ```python
14+ @ttl_cache_decorator(ttl_seconds=60, maxsize=100)
15+ def expensive_function(x: int) -> int:
16+ return x * 2
17+
18+ # First call executes the function
19+ result = expensive_function(5) # Returns 10
20+
21+ # Second call returns cached result
22+ result = expensive_function(5) # Returns 10 (from cache)
23+
24+ # Clear cache manually
25+ expensive_function.clear_cache()
26+ ```
27+ """
28+
29+ def __init__ (self , func : Callable [..., R ], cache : Any , instance : object | None = None ) -> None :
1030 """Initialize the cached function wrapper.
1131
1232 Args:
1333 func: The function to wrap.
1434 cache: The cache instance to use.
35+ instance: The instance this method is bound to (for bound methods).
1536 """
16- self ._func = func
17- self ._cache = cache
37+ self ._func : Callable [..., R ] = func
38+ self ._cache : Any = cache
39+ self ._instance : object | None = instance
1840 # Preserve function metadata
1941 wraps (func )(self )
2042
2143 def __get__ (self , obj : object , objtype : type | None = None ) -> CachedFunction [P , R ]:
22- """Support instance methods by implementing descriptor protocol."""
44+ """Support instance methods by implementing descriptor protocol.
45+
46+ This method caches the bound method in the instance's __dict__ to ensure
47+ identity consistency (obj.method is obj.method returns True).
48+ """
2349 if obj is None :
2450 return self
25- # Return a bound method-like callable
26- from functools import partial
27-
28- bound_call = partial (self .__call__ , obj )
29- # Create a new CachedFunction instance that wraps the bound method
30- # This ensures clear_cache is available on the bound method
31- bound_cached = CachedFunction (bound_call , self ._cache )
32- return bound_cached
3351
34- def __call__ (self , * args : P .args , ** kwargs : P .kwargs ) -> R :
52+ # Cache the bound method in the instance's __dict__ for identity consistency
53+ func_name = getattr (self ._func , "__name__" , "cached_method" )
54+ bound_method_name = f"_cached_{ func_name } "
55+ if not hasattr (obj , bound_method_name ):
56+ # Create a bound CachedFunction that shares the same cache
57+ bound_cached : CachedFunction [P , R ] = CachedFunction (self ._func , self ._cache , instance = obj )
58+ # Store in instance __dict__ to maintain identity
59+ try :
60+ object .__setattr__ (obj , bound_method_name , bound_cached )
61+ except (AttributeError , TypeError ):
62+ # If we can't set the attribute (frozen dataclass, etc.), return a new instance
63+ return CachedFunction (self ._func , self ._cache , instance = obj )
64+
65+ return getattr (obj , bound_method_name )
66+
67+ def __call__ (self , * args : Any , ** kwargs : Any ) -> R :
3568 """Call the cached function.
3669
3770 Args:
@@ -44,47 +77,89 @@ def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
4477 # Create a key based on function name, args, and kwargs
4578 func_name = getattr (self ._func , "__name__" , "unknown" )
4679 key_parts = [func_name ]
47- # Skip first arg if it looks like 'self' (for instance methods)
48- # We check if args[0] has __dict__ which indicates it's likely an instance
49- if args and hasattr (args [0 ], "__dict__" ):
50- key_parts .extend (str (arg ) for arg in args [1 :])
51- else :
52- key_parts .extend (str (arg ) for arg in args )
53- key_parts .extend (f"{ k } :{ v } " for k , v in sorted (kwargs .items ()))
80+
81+ # Use repr() with type information for robust key generation
82+ for arg in args :
83+ key_parts .append (f"{ type (arg ).__name__ } :{ arg !r} " )
84+
85+ # Add keyword arguments
86+ for k , v in sorted (kwargs .items ()):
87+ key_parts .append (f"{ k } ={ type (v ).__name__ } :{ v !r} " )
88+
5489 key = ":" .join (key_parts )
5590
5691 # Check if result is in cache
5792 if key in self ._cache :
5893 return self ._cache [key ]
5994
60- # Call the function and cache the result
61- result = self ._func (* args , ** kwargs )
95+ # Call the function with the instance if this is a bound method
96+ if self ._instance is not None :
97+ result : R = self ._func (self ._instance , * args , ** kwargs )
98+ else :
99+ result = self ._func (* args , ** kwargs )
100+
62101 self ._cache [key ] = result
63102 return result
64103
65104 def clear_cache (self ) -> None :
66- """Clear the cache."""
105+ """Clear the cache.
106+
107+ This clears all cached values for this function. When used with instance methods,
108+ this clears the shared cache for all instances.
109+ """
67110 self ._cache .clear ()
68111
69112
70113def ttl_cache_decorator [** P , R ](
71114 ttl_seconds : int = 300 ,
72115 maxsize : int = 100 ,
73116) -> Callable [[Callable [P , R ]], CachedFunction [P , R ]]:
74- """Decorator that provides a TTL cache for methods.
117+ """Decorator that provides a TTL cache for functions and methods.
118+
119+ The cache is shared across all instances when decorating instance methods.
120+ This is by design to allow efficient caching of expensive operations that
121+ depend only on the method arguments, not the instance state.
75122
76123 Args:
77- ttl_seconds: Time to live in seconds (default: 5 minutes)
78- maxsize: Maximum size of the cache (default: 100)
124+ ttl_seconds: Time to live in seconds (default: 5 minutes).
125+ After this time, cached entries expire and the function is re-executed.
126+ maxsize: Maximum size of the cache (default: 100).
127+ When the cache is full, the least recently used entry is evicted.
79128
80129 Returns:
81- Decorated function with TTL caching
130+ Decorated function with TTL caching and a clear_cache() method.
131+
132+ Example:
133+ ```python
134+ class DataService:
135+ @ttl_cache_decorator(ttl_seconds=60, maxsize=50)
136+ def fetch_data(self, key: str) -> dict:
137+ # Expensive operation
138+ return {"data": key}
139+
140+ service1 = DataService()
141+ service2 = DataService()
142+
143+ # First call executes the function
144+ result1 = service1.fetch_data("key1")
145+
146+ # Second call from different instance returns cached result
147+ result2 = service2.fetch_data("key1") # From cache
148+
149+ # Clear cache manually
150+ service1.fetch_data.clear_cache()
151+ ```
152+
153+ Note:
154+ - Exceptions are not cached; the function will be re-executed on the next call
155+ - None values are cached like any other value
156+ - Cache is shared across all instances of a class (not per-instance)
82157 """
83158 from cachetools import TTLCache
84159
85160 cache : TTLCache = TTLCache (maxsize = maxsize , ttl = ttl_seconds )
86161
87162 def decorator (func : Callable [P , R ]) -> CachedFunction [P , R ]:
88- return CachedFunction (func , cache )
163+ return CachedFunction (func , cache , instance = None )
89164
90165 return decorator
0 commit comments