-
Notifications
You must be signed in to change notification settings - Fork 4
Expand file tree
/
Copy pathbase.py
More file actions
64 lines (48 loc) · 2.52 KB
/
Copy pathbase.py
File metadata and controls
64 lines (48 loc) · 2.52 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
from collections.abc import Callable
from functools import wraps
from typing import Any, TypeVar
from .interfaces import CacheDecoratorInterface
F = TypeVar("F", bound=Callable[..., Any])
class BaseCacheableClass:
def __init__(self, cache_decorator: CacheDecoratorInterface) -> None:
self._cache_decorator = cache_decorator
def wrapped(self, func: F) -> F:
return self._cache_decorator()(func) # type: ignore
@classmethod
def cache(cls, ttl: int | None = None) -> Callable[[F], F]:
# Note: if `ttl` is None, then the cache is stored forever in-memory.
def decorator(func: F) -> F:
@wraps(func)
async def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
if not hasattr(self, "_cache_decorator"):
raise AttributeError("_cache_decorator not found. Did you call super().__init__?")
return await self._cache_decorator(ttl=ttl)(func)(self, *args, **kwargs)
return wrapper # type: ignore
return decorator
@classmethod
def invalidate(cls, target_func_name: str, param_mapping: dict[str, str] | None = None) -> Callable[[F], F]:
"""조건에 따른 캐시 무효화 로직.
target_func: 캐시를 무효화할 대상 함수
param_mapping: 현재 함수의 파라미터와 대상 함수 파라미터 간의 매핑
예: {'user_id': 'customer_id'} -> 현재 함수의 customer_id를 target_func의 user_id로 매핑
"""
def decorator(func: F) -> F:
@wraps(func)
async def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
if not hasattr(self, "_cache_decorator"):
raise AttributeError("_cache_decorator not found. Did you call super().__init__?")
return await self._cache_decorator.invalidate(target_func_name, param_mapping)(func)(
self, *args, **kwargs
)
return wrapper # type: ignore
return decorator
@classmethod
def invalidate_all(cls) -> Callable[[F], F]:
def decorator(func: F) -> F:
@wraps(func)
async def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
if not hasattr(self, "_cache_decorator"):
raise AttributeError("_cache_decorator not found. Did you call super().__init__?")
return await self._cache_decorator.invalidate_all()(func)(self, *args, **kwargs)
return wrapper # type: ignore
return decorator