Coverage for ckanext/udc/search/logic/utils.py: 38%

1from __future__ import annotations 

2 

3import cProfile 

4import pstats 

5import io 

6import time 

7from functools import wraps 

8from typing import Any 

9 

10 

11def profile_func(func): 

12 @wraps(func) 

13 def wrapper(*args, **kwargs): 

14 pr = cProfile.Profile() 

15 pr.enable() 

16 

17 result = func(*args, **kwargs) 

18 

19 pr.disable() 

20 s = io.StringIO() 

21 ps = pstats.Stats(pr, stream=s).sort_stats("cumulative") 

22 ps.print_stats(50) # Top 500 slowest 

23 

24 print(s.getvalue()) # Or write to file 

25 return result 

26 

27 return wrapper 

28 

29 

30def cache_for(seconds, key_func=None): 

31 """Cache function results for ``seconds`` based on an optional key.""" 

32 

33 def decorator(func): 

34 cached: dict[Any, dict[str, Any]] = {} 

35 

36 @wraps(func) 

37 def wrapper(*args, **kwargs): 

38 if key_func: 

39 cache_key = key_func(*args, **kwargs) 

40 else: 

41 if not args and not kwargs: 

42 cache_key = "__default__" 

43 else: 

44 cache_key = repr((args, sorted(kwargs.items()))) 

45 

46 if cache_key is None: 

47 cache_key = "__default__" 

48 

49 now = time.time() 

50 entry = cached.get(cache_key) 

51 if not entry or now - entry["time"] > seconds: 

52 cached[cache_key] = {"time": now, "result": func(*args, **kwargs)} 

53 return cached[cache_key]["result"] 

54 

55 return wrapper 

56 

57 return decorator