9921484972

Committed 13 Jul 2024 04:55PM UTC coverage: 100.0%. Remained the same

Build # 9921484972

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1 from etuzon/version_1.0.x

version 1.0.3

Run Details

90 of 90 branches covered (100.0%)

Branch coverage included in aggregate %.

9 of 26 new or added lines in 1 file covered. (34.62%)

6 existing lines in 2 files now uncovered.

149 of 149 relevant lines covered (100.0%)

13.29 hits per line

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

100.0

/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():
    with __lock:
        global __timer_results
        __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:
                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✔
20	return self.end_date_ms - self.start_date_ms if self.end_date_ms else None	12✔
21
22	def __enter__(self):	15✔
23	self.start_date_ms = TimeUtil.get_current_date_ms()	12✔
24	return self	12✔
25
26	def __exit__(self, exc_type, exc_val, exc_tb):	15✔
27	self.end_date_ms = TimeUtil.get_current_date_ms()	12✔
28
29	self.exception = exc_val	12✔
30
31	if self.exception:	12✔
32	self.stack_trace = traceback.format_exc()	12✔
33
34	def __str__(self) -> str:	15✔
35	return \	12✔
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✔
49	return __max_keys	12✔
50
51
52	def set_max_keys(new_max_keys: int):	12✔
53	global __max_keys
54	__max_keys = new_max_keys	12✔
55
56
57	def get_max_results_per_key() -> int:	12✔
58	return __max_results_per_key	12✔
59
60
61	def set_max_results_per_key(new_max_results_per_key: int):	12✔
62	global __max_results_per_key
63	__max_results_per_key = new_max_results_per_key	12✔
64
65
66	def reset_timer_results():	12✔
UNCOV 67	with __lock:	12✔
68	global __timer_results
UNCOV 69	__timer_results = {}	12✔
70
71
72	def get_timer_results() -> dict:	12✔
73	with __lock:	12✔
74	return __timer_results.copy()	12✔
75
76
77	def method_timer(func, args, *kwargs) -> Timer:	12✔
78	timer_result = Timer()	12✔
79	timer_result.start_date_ms = TimeUtil.get_current_date_ms()	12✔
80
81	try:	12✔
82	timer_result.result = func(args, *kwargs)	12✔
83	except Exception as e:	12✔
84	timer_result.end_date_ms = TimeUtil.get_current_date_ms()	12✔
85	timer_result.exception = e	12✔
86	timer_result.stack_trace = traceback.format_exc()	12✔
87	finally:
88	if timer_result.end_date_ms is None:	12✔
89	timer_result.end_date_ms = TimeUtil.get_current_date_ms()	12✔
90
91	return timer_result	12✔
92
93
94	def timer(is_enabled: bool = True):	12✔
95
96	def decorator(func):	12✔
97
98	def wrapper(args, *kwargs):	12✔
99	timer_result = method_timer(func, args, *kwargs)	12✔
100
101	if is_enabled:	12✔
UNCOV 102	func_file = inspect.getfile(func)	12✔
UNCOV 103	func_qualname = func.__qualname__	12✔
104	key = f'{func_file}:{func_qualname}'	12✔
105
106	with __lock:	12✔
107	if __timer_results.get(key):	12✔
108	if len(__timer_results[key]) >= __max_results_per_key:	12✔
109	__timer_results[key].pop(0)	12✔
110
111	__timer_results[key].append(timer_result)	12✔
112	else:
113	if len(__timer_results) < __max_keys:	12✔
114	__timer_results[key] = [timer_result]	12✔
115
116	if timer_result.exception:	12✔
117	raise timer_result.exception	12✔
118
119	return timer_result.result	12✔
120
121	return wrapper	12✔
122
123	return decorator	12✔

etuzon / python-nrt-time-utils / 9921484972

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