kytos.core.helpers.get_time() - Code Metrics - Inspection of "[Feature] ThreadPoolExecutor for `listen_to` event..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#149)

by Vinicius

created 2021-12-03 21:21 UTC

kytos.core.helpers.get_time() A

↳ Parent: kytos.core.helpers

Complexity

Conditions

Size

Total Lines	30
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	3

Importance

Changes

Metric	Value
cc	3
eloc	7
nop	1
dl	0
loc	30
rs	10
c	0
b	0
f	0
ccs	7
cts	7
cp	1
crap	3

"""Utilities functions used in Kytos."""
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from threading import Thread

from kytos.core.config import KytosConfig

__all__ = ['listen_to', 'now', 'run_on_thread', 'get_time']


# APP_MSG = "[App %s] %s | ID: %02d | R: %02d | P: %02d | F: %s"


def get_thread_pool_max_workers():
    """Get the number of thread pool max workers."""
    return int(KytosConfig().options["daemon"].thread_pool_max_workers)


if get_thread_pool_max_workers():
    executor = ThreadPoolExecutor(max_workers=get_thread_pool_max_workers())


def listen_to(event, *events):
    """Decorate Event Listener methods.

    This decorator was built to be used on NAPPs methods to define which
    type of event the method will handle. With this, we will be able to
    'schedule' the app/method to receive an event when a new event is
    registered on the controller buffers.
    By using the run_on_thread decorator, we also guarantee that the method
    (handler) will be called from inside a new thread, avoiding this method to
    block its caller.

    The decorator will add an attribute to the method called 'events', that
    will be a list of the events that the method will handle.

    The event that will be listened to is always a string, but it can represent
    a regular expression to match against multiple Event Types. All listened
    events are documented in :doc:`/developer/listened_events` section.

    Example of usage:

    .. code-block:: python3

        class MyAppClass(KytosApp):
            @listen_to('kytos/of_core.messages.in')
            def my_handler_of_message_in(self, event):
                # Do stuff here...

            @listen_to('kytos/of_core.messages.out')
            def my_handler_of_message_out(self, event):
                # Do stuff here...

            @listen_to('kytos/of_core.messages.in.ofpt_hello',
                       'kytos/of_core.messages.out.ofpt_hello')
            def my_handler_of_hello_messages(self, event):
                # Do stuff here...

            @listen_to('kytos/of_core.message.*.hello')
            def my_other_handler_of_hello_messages(self, event):
                # Do stuff here...

            @listen_to('kytos/of_core.message.*.hello')
            def my_handler_of_hello_messages(self, event):
                # Do stuff here...

            @listen_to('kytos/of_core.message.*')
            def my_stats_handler_of_any_message(self, event):
                # Do stuff here...
    """
    def thread_decorator(handler):
        """Decorate the handler method.

        Returns:
            A method with an `events` attribute (list of events to be listened)
            and also decorated to run on a new thread.

        """
        @run_on_thread
        def threaded_handler(*args):
            """Decorate the handler to run from a new thread."""
            handler(*args)

        threaded_handler.events = [event]
        threaded_handler.events.extend(events)
        return threaded_handler

    def thread_pool_decorator(handler):
        """Decorate the handler method.

        Returns:
            A method with an `events` attribute (list of events to be listened)
            and also decorated to run on in the thread pool

        """
        def inner(*args):
            """Decorate the handler to run in the thread pool."""
            executor.submit(handler, *args)


        inner.events = [event]
        inner.events.extend(events)
        return inner

    if get_thread_pool_max_workers():
        return thread_pool_decorator
    return thread_decorator


def now(tzone=timezone.utc):
    """Return the current datetime (default to UTC).

    Args:
        tzone (datetime.timezone): Specific time zone used in datetime.

    Returns:
        datetime.datetime.now: Date time with specific time zone.

    """
    return datetime.now(tzone)


def run_on_thread(method):
    """Decorate to run the decorated method inside a new thread.

    Args:
        method (function): function used to run as a new thread.

    Returns:
        Decorated method that will run inside a new thread.
        When the decorated method is called, it will not return the created
        thread to the caller.

    """
    def threaded_method(*args):
        """Ensure the handler method runs inside a new thread."""
        thread = Thread(target=method, args=args)

        # Set daemon mode so that we don't have to wait for these threads
        # to finish when exiting Kytos
        thread.daemon = True
        thread.start()
    return threaded_method


def get_time(data=None):
    """Receive a dictionary or a string and return a datatime instance.

    data = {"year": 2006,
            "month": 11,
            "day": 21,
            "hour": 16,
            "minute": 30 ,
            "second": 00}

    or

    data = "21/11/06 16:30:00"

    2018-04-17T17:13:50Z

    Args:
        data (str, dict): python dict or string to be converted to datetime

    Returns:
        datetime: datetime instance.

    """
    if isinstance(data, str):
        date = datetime.strptime(data, "%Y-%m-%dT%H:%M:%S")
    elif isinstance(data, dict):
        date = datetime(**data)
    else:
        return None
    return date.replace(tzinfo=timezone.utc)


1		"""Utilities functions used in Kytos."""
2	2	from concurrent.futures import ThreadPoolExecutor
3	2	from datetime import datetime, timezone
4	2	from threading import Thread
5
6	2	from kytos.core.config import KytosConfig
7
8	2	__all__ = ['listen_to', 'now', 'run_on_thread', 'get_time']
9
10
11		# APP_MSG = "[App %s] %s \| ID: %02d \| R: %02d \| P: %02d \| F: %s"
12
13
14	2	def get_thread_pool_max_workers():
15		"""Get the number of thread pool max workers."""
16	2	return int(KytosConfig().options["daemon"].thread_pool_max_workers)
17
18
19	2	if get_thread_pool_max_workers():
20	2	executor = ThreadPoolExecutor(max_workers=get_thread_pool_max_workers())
21
22
23	2	def listen_to(event, *events):
24		"""Decorate Event Listener methods.
25
26		This decorator was built to be used on NAPPs methods to define which
27		type of event the method will handle. With this, we will be able to
28		'schedule' the app/method to receive an event when a new event is
29		registered on the controller buffers.
30		By using the run_on_thread decorator, we also guarantee that the method
31		(handler) will be called from inside a new thread, avoiding this method to
32		block its caller.
33
34		The decorator will add an attribute to the method called 'events', that
35		will be a list of the events that the method will handle.
36
37		The event that will be listened to is always a string, but it can represent
38		a regular expression to match against multiple Event Types. All listened
39		events are documented in :doc:`/developer/listened_events` section.
40
41		Example of usage:
42
43		.. code-block:: python3
44
45		class MyAppClass(KytosApp):
46		@listen_to('kytos/of_core.messages.in')
47		def my_handler_of_message_in(self, event):
48		# Do stuff here...
49
50		@listen_to('kytos/of_core.messages.out')
51		def my_handler_of_message_out(self, event):
52		# Do stuff here...
53
54		@listen_to('kytos/of_core.messages.in.ofpt_hello',
55		'kytos/of_core.messages.out.ofpt_hello')
56		def my_handler_of_hello_messages(self, event):
57		# Do stuff here...
58
59		@listen_to('kytos/of_core.message.*.hello')
60		def my_other_handler_of_hello_messages(self, event):
61		# Do stuff here...
62
63		@listen_to('kytos/of_core.message.*.hello')
64		def my_handler_of_hello_messages(self, event):
65		# Do stuff here...
66
67		@listen_to('kytos/of_core.message.*')
68		def my_stats_handler_of_any_message(self, event):
69		# Do stuff here...
70		"""
71		def thread_decorator(handler):
72		"""Decorate the handler method.
73
74		Returns:
75		A method with an `events` attribute (list of events to be listened)
76		and also decorated to run on a new thread.
77
78		"""
79		@run_on_thread
80		def threaded_handler(*args):
81		"""Decorate the handler to run from a new thread."""
82		handler(*args)
83
84		threaded_handler.events = [event]
85		threaded_handler.events.extend(events)
86		return threaded_handler
87
88		def thread_pool_decorator(handler):
89		"""Decorate the handler method.
90
91		Returns:
92		A method with an `events` attribute (list of events to be listened)
93		and also decorated to run on in the thread pool
94
95		"""
96		def inner(*args):
97		"""Decorate the handler to run in the thread pool."""
98		executor.submit(handler, *args)
		0 ignored issues – show introduced 2021-12-03 21:27 UTC by Report Bug Copy Issue Report The variable `executor` does not seem to be defined in case `get_thread_pool_max_workers()` on line `19` is `False`. Are you sure this can never be the case? Loading history...
99
100		inner.events = [event]
101		inner.events.extend(events)
102		return inner
103
104		if get_thread_pool_max_workers():
105		return thread_pool_decorator
106		return thread_decorator
107
108
109	2	def now(tzone=timezone.utc):
110		"""Return the current datetime (default to UTC).
111
112		Args:
113		tzone (datetime.timezone): Specific time zone used in datetime.
114
115		Returns:
116		datetime.datetime.now: Date time with specific time zone.
117
118		"""
119	2	return datetime.now(tzone)
120
121
122	2	def run_on_thread(method):
123		"""Decorate to run the decorated method inside a new thread.
124
125		Args:
126		method (function): function used to run as a new thread.
127
128		Returns:
129		Decorated method that will run inside a new thread.
130		When the decorated method is called, it will not return the created
131		thread to the caller.
132
133		"""
134	2	def threaded_method(*args):
135		"""Ensure the handler method runs inside a new thread."""
136	2	thread = Thread(target=method, args=args)
137
138		# Set daemon mode so that we don't have to wait for these threads
139		# to finish when exiting Kytos
140	2	thread.daemon = True
141	2	thread.start()
142	2	return threaded_method
143
144
145	2	def get_time(data=None):
146		"""Receive a dictionary or a string and return a datatime instance.
147
148		data = {"year": 2006,
149		"month": 11,
150		"day": 21,
151		"hour": 16,
152		"minute": 30 ,
153		"second": 00}
154
155		or
156
157		data = "21/11/06 16:30:00"
158
159		2018-04-17T17:13:50Z
160
161		Args:
162		data (str, dict): python dict or string to be converted to datetime
163
164		Returns:
165		datetime: datetime instance.
166
167		"""
168	2	if isinstance(data, str):
169	2	date = datetime.strptime(data, "%Y-%m-%dT%H:%M:%S")
170	2	elif isinstance(data, dict):
171	2	date = datetime(**data)
172		else:
173	2	return None
174		return date.replace(tzinfo=timezone.utc)
175

kytos-ng / kytos

Pull Request — master (#149)

kytos.core.helpers.get_time() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like