kytos.core.helpers.now() - Code Metrics - Inspection of "[Feature] Dead letter structure for unhandled exce..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#204)

by Vinicius

created 2022-05-06 14:52 UTC

kytos.core.helpers.now() A

↳ Parent: kytos.core.helpers

Complexity

Conditions

Size

Total Lines	11
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	11
rs	10
c	0
b	0
f	0
ccs	2
cts	2
cp	1
crap	1

"""Utilities functions used in Kytos."""
import logging
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']

LOG = logging.getLogger(__name__)

# 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)


# pylint: disable=invalid-name
executor = None
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 done_callback(future):
            """Done callback."""
            if not future.exception():
                _ = future.result()
                return

            exc_str = f"{type(future.exception())}: {future.exception()}"
            args = (
                getattr(future, "__args")
                if hasattr(future, "__args")
                else tuple()
            )
            LOG.error(f"listen_to handler: {handler}, "
                      f"args: {args}, exception: {exc_str}")
            if len(args) > 1 and hasattr(args[0], "controller"):
                cls, kytos_event = args[0], args[1]
                cls.controller.dead_letter.add_event(kytos_event)

        def inner(*args):
            """Decorate the handler to run in the thread pool."""
            future = executor.submit(handler, *args)
            setattr(future, "__args", args)
            future.add_done_callback(done_callback)

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

kytos-ng / kytos

Pull Request — master (#204)

kytos.core.helpers.now() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like