9627818263

Committed 22 Jun 2024 07:32PM UTC coverage: 43.836% (-19.9%) from 63.736%

Build Type

push

github

Committed by Eyal Tuzon

Commit Message

version 1.0.1

Run Details

24 of 84 branches covered (28.57%)

Branch coverage included in aggregate %.

34 of 76 new or added lines in 1 file covered. (44.74%)

72 of 135 relevant lines covered (53.33%)

7.69 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

29.69

/nrt_time_utils/timer.py

import inspect
import traceback
from dataclasses import dataclass
from threading import Lock
from typing import Optional

from nrt_time_utils.time_utils import TimeUtil


@dataclass
class Timer:
    start_date_ms: Optional[int] = None
    end_date_ms: Optional[int] = None
    result: any = None
    exception: Optional[Exception] = None
    stack_trace: Optional[str] = None

    @property
    def execution_time_ms(self) -> Optional[int]:
        return self.end_date_ms - self.start_date_ms if self.end_date_ms else None

    def __enter__(self):
        self.start_date_ms = TimeUtil.get_current_date_ms()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.end_date_ms = TimeUtil.get_current_date_ms()

        self.exception = exc_val

        if self.exception:
            self.stack_trace = traceback.format_exc()

    def __str__(self) -> str:
        return \
            f'Start date ms = {self.start_date_ms}.'\
            f' End date ms = {self.end_date_ms}.'\
            f' Execution time ms = {self.execution_time_ms}'


__max_keys: int = 100
__max_results_per_key: int = 10000

__lock = Lock()
__timer_results: dict[str, list[Timer]] = {}


def get_max_keys() -> int:
    return __max_keys


def set_max_keys(new_max_keys: int):
    global __max_keys
    __max_keys = new_max_keys


def get_max_results_per_key() -> int:
    return __max_results_per_key


def set_max_results_per_key(new_max_results_per_key: int):
    global __max_results_per_key
    __max_results_per_key = new_max_results_per_key


def reset_timer_results():
    global __timer_results

    with __lock:
        __timer_results = {}


def get_timer_results() -> dict:
    with __lock:
        return __timer_results.copy()


def method_timer(func, *args, **kwargs) -> Timer:
    timer_result = Timer()
    timer_result.start_date_ms = TimeUtil.get_current_date_ms()

    try:
        timer_result.result = func(*args, **kwargs)
    except Exception as e:
        timer_result.end_date_ms = TimeUtil.get_current_date_ms()
        timer_result.exception = e
        timer_result.stack_trace = traceback.format_exc()
    finally:
        if timer_result.end_date_ms is None:
            timer_result.end_date_ms = TimeUtil.get_current_date_ms()

    return timer_result


def timer(is_enabled: bool = True):

    def decorator(func):

        def wrapper(*args, **kwargs):
            timer_result = method_timer(func, *args, **kwargs)

            if is_enabled:
                global __timer_results

                func_file = inspect.getfile(func)
                func_qualname = func.__qualname__
                key = f'{func_file}:{func_qualname}'

                with __lock:
                    if __timer_results.get(key):
                        if len(__timer_results[key]) >= __max_results_per_key:
                            __timer_results[key].pop(0)

                        __timer_results[key].append(timer_result)
                    else:
                        if len(__timer_results) < __max_keys:
                            __timer_results[key] = [timer_result]

            if timer_result.exception:
                raise timer_result.exception

            return timer_result.result

        return wrapper

    return decorator

1	import inspect	15✔
2	import traceback	15✔
3	from dataclasses import dataclass	15✔
4	from threading import Lock	15✔
5	from typing import Optional	15✔
6
7	from nrt_time_utils.time_utils import TimeUtil	15✔
8
9
10	@dataclass	15✔
11	class Timer:	15✔
12	start_date_ms: Optional[int] = None	15✔
13	end_date_ms: Optional[int] = None	15✔
14	result: any = None	15✔
15	exception: Optional[Exception] = None	15✔
16	stack_trace: Optional[str] = None	15✔
17
18	@property	15✔
19	def execution_time_ms(self) -> Optional[int]:	15✔
NEW 20	return self.end_date_ms - self.start_date_ms if self.end_date_ms else None	×
21
22	def __enter__(self):	15✔
NEW 23	self.start_date_ms = TimeUtil.get_current_date_ms()	×
NEW 24	return self	×
25
26	def __exit__(self, exc_type, exc_val, exc_tb):	15✔
NEW 27	self.end_date_ms = TimeUtil.get_current_date_ms()	×
28
NEW 29	self.exception = exc_val	×
30
NEW 31	if self.exception:	×
NEW 32	self.stack_trace = traceback.format_exc()	×
33
34	def __str__(self) -> str:	15✔
NEW 35	return \	×
36	f'Start date ms = {self.start_date_ms}.'\
37	f' End date ms = {self.end_date_ms}.'\
38	f' Execution time ms = {self.execution_time_ms}'
39
40
41	__max_keys: int = 100	15✔
42	__max_results_per_key: int = 10000	15✔
43
44	__lock = Lock()	15✔
45	__timer_results: dict[str, list[Timer]] = {}	15✔
46
47
48	def get_max_keys() -> int:	12✔
NEW 49	return __max_keys	×
50
51
52	def set_max_keys(new_max_keys: int):	12✔
53	global __max_keys
NEW 54	__max_keys = new_max_keys	×
55
56
57	def get_max_results_per_key() -> int:	12✔
NEW 58	return __max_results_per_key	×
59
60
61	def set_max_results_per_key(new_max_results_per_key: int):	12✔
62	global __max_results_per_key
NEW 63	__max_results_per_key = new_max_results_per_key	×
64
65
66	def reset_timer_results():	12✔
67	global __timer_results
68
NEW 69	with __lock:	×
NEW 70	__timer_results = {}	×
71
72
73	def get_timer_results() -> dict:	12✔
NEW 74	with __lock:	×
NEW 75	return __timer_results.copy()	×
76
77
78	def method_timer(func, args, *kwargs) -> Timer:	12✔
NEW 79	timer_result = Timer()	×
NEW 80	timer_result.start_date_ms = TimeUtil.get_current_date_ms()	×
81
NEW 82	try:	×
NEW 83	timer_result.result = func(args, *kwargs)	×
NEW 84	except Exception as e:	×
NEW 85	timer_result.end_date_ms = TimeUtil.get_current_date_ms()	×
NEW 86	timer_result.exception = e	×
NEW 87	timer_result.stack_trace = traceback.format_exc()	×
88	finally:
NEW 89	if timer_result.end_date_ms is None:	×
NEW 90	timer_result.end_date_ms = TimeUtil.get_current_date_ms()	×
91
NEW 92	return timer_result	×
93
94
95	def timer(is_enabled: bool = True):	12✔
96
97	def decorator(func):	12✔
98
99	def wrapper(args, *kwargs):	12✔
NEW 100	timer_result = method_timer(func, args, *kwargs)	×
101
NEW 102	if is_enabled:	×
103	global __timer_results
104
NEW 105	func_file = inspect.getfile(func)	×
NEW 106	func_qualname = func.__qualname__	×
NEW 107	key = f'{func_file}:{func_qualname}'	×
108
NEW 109	with __lock:	×
NEW 110	if __timer_results.get(key):	×
NEW 111	if len(__timer_results[key]) >= __max_results_per_key:	×
NEW 112	__timer_results[key].pop(0)	×
113
NEW 114	__timer_results[key].append(timer_result)	×
115	else:
NEW 116	if len(__timer_results) < __max_keys:	×
NEW 117	__timer_results[key] = [timer_result]	×
118
NEW 119	if timer_result.exception:	×
NEW 120	raise timer_result.exception	×
121
NEW 122	return timer_result.result	×
123
124	return wrapper	12✔
125
126	return decorator	12✔

etuzon / python-nrt-time-utils / 9627818263

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous