Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/prometheus_client/context_managers.py: 41%

1from timeit import default_timer 

2from types import TracebackType 

3from typing import ( 

4 Any, Callable, Literal, Optional, Tuple, Type, TYPE_CHECKING, TypeVar, 

5 Union, 

6) 

7 

8from .decorator import decorate 

9 

10if TYPE_CHECKING: 

11 from . import Counter 

12 F = TypeVar("F", bound=Callable[..., Any]) 

13 

14 

15class ExceptionCounter: 

16 def __init__(self, counter: "Counter", exception: Union[Type[BaseException], Tuple[Type[BaseException], ...]]) -> None: 

17 self._counter = counter 

18 self._exception = exception 

19 

20 def __enter__(self) -> None: 

21 pass 

22 

23 def __exit__(self, typ: Optional[Type[BaseException]], value: Optional[BaseException], traceback: Optional[TracebackType]) -> Literal[False]: 

24 if isinstance(value, self._exception): 

25 self._counter.inc() 

26 return False 

27 

28 def __call__(self, f: "F") -> "F": 

29 def wrapped(func, *args, **kwargs): 

30 with self: 

31 return func(*args, **kwargs) 

32 

33 return decorate(f, wrapped) 

34 

35 

36class InprogressTracker: 

37 def __init__(self, gauge): 

38 self._gauge = gauge 

39 

40 def __enter__(self): 

41 self._gauge.inc() 

42 

43 def __exit__(self, typ, value, traceback): 

44 self._gauge.dec() 

45 

46 def __call__(self, f: "F") -> "F": 

47 def wrapped(func, *args, **kwargs): 

48 with self: 

49 return func(*args, **kwargs) 

50 

51 return decorate(f, wrapped) 

52 

53 

54class Timer: 

55 def __init__(self, metric, callback_name): 

56 self._metric = metric 

57 self._callback_name = callback_name 

58 

59 def _new_timer(self): 

60 return self.__class__(self._metric, self._callback_name) 

61 

62 def __enter__(self): 

63 self._start = default_timer() 

64 return self 

65 

66 def __exit__(self, typ, value, traceback): 

67 # Time can go backwards. 

68 duration = max(default_timer() - self._start, 0) 

69 callback = getattr(self._metric, self._callback_name) 

70 callback(duration) 

71 

72 def labels(self, *args, **kw): 

73 self._metric = self._metric.labels(*args, **kw) 

74 

75 def __call__(self, f: "F") -> "F": 

76 def wrapped(func, *args, **kwargs): 

77 # Obtaining new instance of timer every time 

78 # ensures thread safety and reentrancy. 

79 with self._new_timer(): 

80 return func(*args, **kwargs) 

81 

82 return decorate(f, wrapped)