Coverage for postrfp/shared/utils.py: 100%

1import io 

2import time 

3import pstats 

4import logging 

5import cProfile 

6import contextlib 

7from enum import Enum 

8from decimal import Decimal 

9 

10from pydantic import AnyHttpUrl 

11 

12 

13bench_log = logging.getLogger("postrfp.benchmark") 

14log = logging.getLogger(__name__) 

15 

16 

17class benchmark(object): # pragma: no cover 

18 """ 

19 Context manager for measuring and logging execution time of code blocks. 

20 

21 This utility logs the execution time with varying log levels based on how long 

22 the operation takes. Longer execution times result in higher severity log levels. 

23 

24 Parameters: 

25 name (str): Descriptive name for the operation being timed, included in log output 

26 sensitivity (float, optional): Multiplier that adjusts timing thresholds. 

27 Higher values make the benchmark more sensitive to timing differences, 

28 triggering higher log levels at shorter durations. Defaults to 1. 

29 

30 Log level thresholds: 

31 - DEBUG: time < (0.1 / sensitivity) seconds 

32 - INFO: time < (0.5 / sensitivity) seconds 

33 - WARN: time < (2.0 / sensitivity) seconds 

34 - ERROR: time ≥ (2.0 / sensitivity) seconds 

35 

36 Example: 

37 >>> with benchmark("database query", sensitivity=2): 

38 ... results = db.execute_complex_query() 

39 

40 # For a query taking 0.8 seconds with sensitivity=2: 

41 # Will log at WARN level: "database query : 0.800 seconds" 

42 """ 

43 

44 def __init__(self, name, sensitivity=1): 

45 self.name = name 

46 self.sensitivity = sensitivity 

47 

48 def __enter__(self): 

49 self.start = time.time() 

50 

51 def __exit__(self, _ty, val, _tb): 

52 elapsed = time.time() - self.start 

53 if elapsed < (0.1 / self.sensitivity): 

54 level = logging.DEBUG 

55 elif elapsed < (0.5 / self.sensitivity): 

56 level = logging.INFO 

57 elif elapsed < (2 / self.sensitivity): 

58 level = logging.WARN 

59 else: 

60 level = logging.ERROR 

61 bench_log.log(level, "%s : %0.3f seconds", self.name, elapsed) 

62 return False 

63 

64 

65@contextlib.contextmanager # pragma: no cover 

66def profiled(): 

67 pr = cProfile.Profile() 

68 pr.enable() 

69 yield pr 

70 pr.disable() 

71 s = io.StringIO() 

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

73 # ps.print_stats(1000, "postrfp") 

74 ps.print_stats() 

75 # uncomment this to see who's calling what 

76 # ps.print_callers() 

77 print(s.getvalue()) 

78 

79 

80def json_default(obj): 

81 if isinstance(obj, set): 

82 return list(obj) 

83 elif isinstance(obj, Decimal): 

84 return float(obj) 

85 elif hasattr(obj, "as_dict"): 

86 return obj.as_dict() 

87 elif isinstance(obj, Enum): 

88 return obj.value 

89 elif isinstance(obj, AnyHttpUrl): 

90 return str(obj) 

91 else: 

92 raise TypeError("Cannot serialise %s to json." % type(obj))