kytos.core.helpers.load_spec() - Code Metrics - Inspection of "Merge pull request #352 from kytos-ng/feat/openapi..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( ad4b11...eeb2a1 )

by Vinicius

created 2023-03-30 15:32 UTC

kytos.core.helpers.load_spec() A

↳ Parent: kytos.core.helpers

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	1

Importance

Changes

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

"""Utilities functions used in Kytos."""
import functools
import logging
import traceback
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from pathlib import Path
from threading import Thread

from flask import request
from openapi_core.contrib.flask import FlaskOpenAPIRequest
from openapi_core.spec.shortcuts import create_spec
from openapi_core.validation.request import openapi_request_validator
from openapi_spec_validator import validate_spec
from openapi_spec_validator.readers import read_from_filename
from werkzeug.exceptions import BadRequest, UnsupportedMediaType

from kytos.core.apm import ElasticAPM
from kytos.core.config import KytosConfig

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

LOG = logging.getLogger(__name__)


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


def get_apm_name():
    """Get apm backend name."""
    return KytosConfig().options["daemon"].apm


# pylint: disable=invalid-name
executors = {name: ThreadPoolExecutor(max_workers=max_workers,
                                      thread_name_prefix=f"thread_pool_{name}")
             for name, max_workers in get_thread_pool_max_workers().items()}


def listen_to(event, *events, pool=None):
    """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.

    The pool option gives you control on which ThreadPoolExecutor that will
    execute the decorated handler. This knob is meant for prioritization,
    allowing executions on different pools depending on the handler primary
    responsibility and importance to avoid potential scheduling starvation.

    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...

            @listen_to("some_db_oriented_event", pool="db")
            def db_update(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

    # pylint: disable=broad-except
    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

        # pylint: disable=unused-argument
        def handler_context(*args, **kwargs):

            """Handler's context for ThreadPool."""
            cls, kytos_event = args[0], args[1]
            try:
                result = handler(*args)
            except Exception:
                result = None
                traceback_str = traceback.format_exc().replace("\n", ", ")
                LOG.error(f"listen_to handler: {handler}, "
                          f"args: {args} traceback: {traceback_str}")
                if hasattr(cls, "controller"):
                    cls.controller.dead_letter.add_event(kytos_event)
            return result

        def handler_context_apm(*args, apm_client=None):

            """Handler's context for ThreadPool APM instrumentation."""
            cls, kytos_event = args[0], args[1]
            trace_parent = kytos_event.trace_parent
            tx_type = "kytos_event"
            tx = apm_client.begin_transaction(transaction_type=tx_type,
                                              trace_parent=trace_parent)
            kytos_event.trace_parent = tx.trace_parent
            tx.name = f"{kytos_event.name}@{cls.napp_id}"
            try:
                result = handler(*args)
                tx.result = result
            except Exception as exc:
                result = None
                traceback_str = traceback.format_exc().replace("\n", ", ")
                LOG.error(f"listen_to handler: {handler}, "
                          f"args: {args} traceback: {traceback_str}")
                if hasattr(cls, "controller"):
                    cls.controller.dead_letter.add_event(kytos_event)
                apm_client.capture_exception(
                    exc_info=(type(exc), exc, exc.__traceback__),
                    context={"args": args},
                    handled=False,
                )
            tx.end()
            apm_client.tracer.queue_func("transaction", tx.to_dict())
            return result

        handler_func, kwargs = handler_context, {}
        if get_apm_name() == "es":
            handler_func = handler_context_apm
            kwargs = dict(apm_client=ElasticAPM.get_client())

        def get_executor(pool, event, default_pool="app"):
            """Get executor."""
            if pool:
                return executors[pool]
            if not event:
                return executors[default_pool]

            if event.name.startswith("kytos/of_core") and "sb" in executors:
                return executors["sb"]
            core_of = "kytos/core.openflow"
            if event.name.startswith(core_of) and "sb" in executors:
                return executors["sb"]
            if event.name.startswith("kytos.storehouse") and "db" in executors:
                return executors["db"]
            return executors[default_pool]

        def inner(*args):
            """Decorate the handler to run in the thread pool."""
            event = args[1] if len(args) > 1 else None
            executor = get_executor(pool, event)
            future = executor.submit(handler_func, *args, **kwargs)
            future.add_done_callback(done_callback)

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

    if executors:
        return thread_pool_decorator
    return thread_decorator


def alisten_to(event, *events):
    """Decorate async subscribing methods."""

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

        Returns:
            A method with an `events` attribute (list of events to be listened)
            and also decorated as an asyncio task.
        """
        # pylint: disable=unused-argument,broad-except
        async def handler_context(*args, **kwargs):

            """Async handler's execution context."""
            cls, kytos_event = args[0], args[1]
            try:
                result = await handler(*args)
            except Exception:
                result = None
                traceback_str = traceback.format_exc().replace("\n", ", ")
                LOG.error(f"alisten_to handler: {handler}, "
                          f"args: {args} traceback: {traceback_str}")
                if hasattr(cls, "controller"):
                    cls.controller.dead_letter.add_event(kytos_event)
            return result

        async def handler_context_apm(*args, apm_client=None):

            """Async handler's execution context with APM instrumentation."""
            cls, kytos_event = args[0], args[1]
            trace_parent = kytos_event.trace_parent
            tx_type = "kytos_event"
            tx = apm_client.begin_transaction(transaction_type=tx_type,
                                              trace_parent=trace_parent)
            kytos_event.trace_parent = tx.trace_parent
            tx.name = f"{kytos_event.name}@{cls.napp_id}"
            try:
                result = await handler(*args)
                tx.result = result
            except Exception as exc:
                result = None
                traceback_str = traceback.format_exc().replace("\n", ", ")
                LOG.error(f"alisten_to handler: {handler}, "
                          f"args: {args} traceback: {traceback_str}")
                if hasattr(cls, "controller"):
                    cls.controller.dead_letter.add_event(kytos_event)
                apm_client.capture_exception(
                    exc_info=(type(exc), exc, exc.__traceback__),
                    context={"args": args},
                    handled=False,
                )
            tx.end()
            apm_client.tracer.queue_func("transaction", tx.to_dict())
            return result

        handler_func, kwargs = handler_context, {}
        if get_apm_name() == "es":
            handler_func = handler_context_apm
            kwargs = dict(apm_client=ElasticAPM.get_client())

        async def inner(*args):
            """Inner decorated with events attribute."""
            return await handler_func(*args, **kwargs)
        inner.events = [event]
        inner.events.extend(events)
        return inner

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


def _read_from_filename(yml_file_path: Path) -> dict:
    """Read from yml filename."""
    spec_dict, _ = read_from_filename(yml_file_path)
    return spec_dict


def load_spec(yml_file_path: Path):
    """Load and validate spec object given a yml file path."""
    spec_dict = _read_from_filename(yml_file_path)
    validate_spec(spec_dict)
    return create_spec(spec_dict)


def validate_openapi(spec):
    """Decorator to validate a REST endpoint input.

    Uses the schema defined in the openapi.yml file
    to validate.
    """

    def validate_decorator(func):
        @functools.wraps(func)
        def wrapper_validate(*args, **kwargs):
            try:
                data = request.get_json()
            except BadRequest:
                result = "The request body is not a well-formed JSON."
                raise BadRequest(result) from BadRequest
            if data is None:
                result = "The request body mimetype is not application/json."
                raise UnsupportedMediaType(result)

            openapi_request = FlaskOpenAPIRequest(request)
            result = openapi_request_validator.validate(spec, openapi_request)
            if result.errors:
                error_response = (
                    "The request body contains invalid API data."
                )
                errors = result.errors[0]
                if hasattr(errors, "schema_errors"):
                    schema_errors = errors.schema_errors[0]
                    error_log = {
                        "error_message": schema_errors.message,
                        "error_validator": schema_errors.validator,
                        "error_validator_value": schema_errors.validator_value,
                        "error_path": list(schema_errors.path),
                        "error_schema": schema_errors.schema,
                        "error_schema_path": list(schema_errors.schema_path),
                    }
                    LOG.debug(f"Invalid request (API schema): {error_log}")
                    error_response += f" {schema_errors.message} for field"
                    error_response += (
                        f" {'/'.join(map(str,schema_errors.path))}."
                    )
                raise BadRequest(error_response) from BadRequest
            return func(*args, data=data, **kwargs)

        return wrapper_validate

    return validate_decorator


1			"""Utilities functions used in Kytos."""
2	1		import functools
3	1		import logging
4	1		import traceback
5	1		from concurrent.futures import ThreadPoolExecutor
6	1		from datetime import datetime, timezone
7	1		from pathlib import Path
8	1		from threading import Thread
9
10	1		from flask import request
11	1		from openapi_core.contrib.flask import FlaskOpenAPIRequest
12	1		from openapi_core.spec.shortcuts import create_spec
13	1		from openapi_core.validation.request import openapi_request_validator
14	1		from openapi_spec_validator import validate_spec
15	1		from openapi_spec_validator.readers import read_from_filename
16	1		from werkzeug.exceptions import BadRequest, UnsupportedMediaType
17
18	1		from kytos.core.apm import ElasticAPM
19	1		from kytos.core.config import KytosConfig
20
21	1		__all__ = ['listen_to', 'now', 'run_on_thread', 'get_time']
22
23	1		LOG = logging.getLogger(__name__)
24
25
26	1		def get_thread_pool_max_workers():
27			"""Get the number of thread pool max workers."""
28	1		return KytosConfig().options["daemon"].thread_pool_max_workers
29
30
31	1		def get_apm_name():
32			"""Get apm backend name."""
33	1		return KytosConfig().options["daemon"].apm
34
35
36			# pylint: disable=invalid-name
37	1		executors = {name: ThreadPoolExecutor(max_workers=max_workers,
38			thread_name_prefix=f"thread_pool_{name}")
39			for name, max_workers in get_thread_pool_max_workers().items()}
40
41
42	1		def listen_to(event, *events, pool=None):
43			"""Decorate Event Listener methods.
44
45			This decorator was built to be used on NAPPs methods to define which
46			type of event the method will handle. With this, we will be able to
47			'schedule' the app/method to receive an event when a new event is
48			registered on the controller buffers.
49			By using the run_on_thread decorator, we also guarantee that the method
50			(handler) will be called from inside a new thread, avoiding this method to
51			block its caller.
52
53			The decorator will add an attribute to the method called 'events', that
54			will be a list of the events that the method will handle.
55
56			The event that will be listened to is always a string, but it can represent
57			a regular expression to match against multiple Event Types. All listened
58			events are documented in :doc:`/developer/listened_events` section.
59
60			The pool option gives you control on which ThreadPoolExecutor that will
61			execute the decorated handler. This knob is meant for prioritization,
62			allowing executions on different pools depending on the handler primary
63			responsibility and importance to avoid potential scheduling starvation.
64
65			Example of usage:
66
67			.. code-block:: python3
68
69			class MyAppClass(KytosApp):
70			@listen_to('kytos/of_core.messages.in')
71			def my_handler_of_message_in(self, event):
72			# Do stuff here...
73
74			@listen_to('kytos/of_core.messages.out')
75			def my_handler_of_message_out(self, event):
76			# Do stuff here...
77
78			@listen_to('kytos/of_core.messages.in.ofpt_hello',
79			'kytos/of_core.messages.out.ofpt_hello')
80			def my_handler_of_hello_messages(self, event):
81			# Do stuff here...
82
83			@listen_to('kytos/of_core.message.*.hello')
84			def my_other_handler_of_hello_messages(self, event):
85			# Do stuff here...
86
87			@listen_to('kytos/of_core.message.*.hello')
88			def my_handler_of_hello_messages(self, event):
89			# Do stuff here...
90
91			@listen_to('kytos/of_core.message.*')
92			def my_stats_handler_of_any_message(self, event):
93			# Do stuff here...
94
95			@listen_to("some_db_oriented_event", pool="db")
96			def db_update(self, event):
97			# Do stuff here...
98			"""
99	1		def thread_decorator(handler):
100			"""Decorate the handler method.
101
102			Returns:
103			A method with an `events` attribute (list of events to be listened)
104			and also decorated to run on a new thread.
105
106			"""
107			@run_on_thread
108			def threaded_handler(*args):
109			"""Decorate the handler to run from a new thread."""
110			handler(*args)
111
112			threaded_handler.events = [event]
113			threaded_handler.events.extend(events)
114			return threaded_handler
115
116			# pylint: disable=broad-except
117	1		def thread_pool_decorator(handler):
118			"""Decorate the handler method.
119
120			Returns:
121			A method with an `events` attribute (list of events to be listened)
122			and also decorated to run on in the thread pool
123
124			"""
125	1		def done_callback(future):
126			"""Done callback."""
127			if not future.exception():
128			_ = future.result()
129			return
130
131			# pylint: disable=unused-argument
132	1	View Code Duplication	def handler_context(args, *kwargs):
			0 ignored issues – show Duplication introduced 2022-07-18 14:28 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
133			"""Handler's context for ThreadPool."""
134			cls, kytos_event = args[0], args[1]
135			try:
136			result = handler(*args)
137			except Exception:
138			result = None
139			traceback_str = traceback.format_exc().replace("\n", ", ")
140			LOG.error(f"listen_to handler: {handler}, "
141			f"args: {args} traceback: {traceback_str}")
142			if hasattr(cls, "controller"):
143			cls.controller.dead_letter.add_event(kytos_event)
144			return result
145
146	1	View Code Duplication	def handler_context_apm(*args, apm_client=None):
			0 ignored issues – show Duplication introduced 2022-06-22 15:56 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
147			"""Handler's context for ThreadPool APM instrumentation."""
148			cls, kytos_event = args[0], args[1]
149			trace_parent = kytos_event.trace_parent
150			tx_type = "kytos_event"
151			tx = apm_client.begin_transaction(transaction_type=tx_type,
152			trace_parent=trace_parent)
153			kytos_event.trace_parent = tx.trace_parent
154			tx.name = f"{kytos_event.name}@{cls.napp_id}"
155			try:
156			result = handler(*args)
157			tx.result = result
158			except Exception as exc:
159			result = None
160			traceback_str = traceback.format_exc().replace("\n", ", ")
161			LOG.error(f"listen_to handler: {handler}, "
162			f"args: {args} traceback: {traceback_str}")
163			if hasattr(cls, "controller"):
164			cls.controller.dead_letter.add_event(kytos_event)
165			apm_client.capture_exception(
166			exc_info=(type(exc), exc, exc.__traceback__),
167			context={"args": args},
168			handled=False,
169			)
170			tx.end()
171			apm_client.tracer.queue_func("transaction", tx.to_dict())
172			return result
173
174	1		handler_func, kwargs = handler_context, {}
175	1		if get_apm_name() == "es":
176			handler_func = handler_context_apm
177			kwargs = dict(apm_client=ElasticAPM.get_client())
178
179	1		def get_executor(pool, event, default_pool="app"):
180			"""Get executor."""
181	1		if pool:
182			return executors[pool]
183	1		if not event:
184	1		return executors[default_pool]
185
186			if event.name.startswith("kytos/of_core") and "sb" in executors:
187			return executors["sb"]
188			core_of = "kytos/core.openflow"
189			if event.name.startswith(core_of) and "sb" in executors:
190			return executors["sb"]
191			if event.name.startswith("kytos.storehouse") and "db" in executors:
192			return executors["db"]
193			return executors[default_pool]
194
195	1		def inner(*args):
196			"""Decorate the handler to run in the thread pool."""
197	1		event = args[1] if len(args) > 1 else None
198	1		executor = get_executor(pool, event)
199	1		future = executor.submit(handler_func, args, *kwargs)
200	1		future.add_done_callback(done_callback)
201
202	1		inner.events = [event]
203	1		inner.events.extend(events)
204	1		return inner
205
206	1		if executors:
207	1		return thread_pool_decorator
208			return thread_decorator
209
210
211	1		def alisten_to(event, *events):
212			"""Decorate async subscribing methods."""
213
214	1		def decorator(handler):
215			"""Decorate the handler method.
216
217			Returns:
218			A method with an `events` attribute (list of events to be listened)
219			and also decorated as an asyncio task.
220			"""
221			# pylint: disable=unused-argument,broad-except
222	1	View Code Duplication	async def handler_context(args, *kwargs):
			0 ignored issues – show Duplication introduced 2022-07-18 14:28 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
223			"""Async handler's execution context."""
224	1		cls, kytos_event = args[0], args[1]
225	1		try:
226	1		result = await handler(*args)
227			except Exception:
228			result = None
229			traceback_str = traceback.format_exc().replace("\n", ", ")
230			LOG.error(f"alisten_to handler: {handler}, "
231			f"args: {args} traceback: {traceback_str}")
232			if hasattr(cls, "controller"):
233			cls.controller.dead_letter.add_event(kytos_event)
234	1		return result
235
236	1	View Code Duplication	async def handler_context_apm(*args, apm_client=None):
			0 ignored issues – show Duplication introduced 2022-06-22 15:56 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
237			"""Async handler's execution context with APM instrumentation."""
238			cls, kytos_event = args[0], args[1]
239			trace_parent = kytos_event.trace_parent
240			tx_type = "kytos_event"
241			tx = apm_client.begin_transaction(transaction_type=tx_type,
242			trace_parent=trace_parent)
243			kytos_event.trace_parent = tx.trace_parent
244			tx.name = f"{kytos_event.name}@{cls.napp_id}"
245			try:
246			result = await handler(*args)
247			tx.result = result
248			except Exception as exc:
249			result = None
250			traceback_str = traceback.format_exc().replace("\n", ", ")
251			LOG.error(f"alisten_to handler: {handler}, "
252			f"args: {args} traceback: {traceback_str}")
253			if hasattr(cls, "controller"):
254			cls.controller.dead_letter.add_event(kytos_event)
255			apm_client.capture_exception(
256			exc_info=(type(exc), exc, exc.__traceback__),
257			context={"args": args},
258			handled=False,
259			)
260			tx.end()
261			apm_client.tracer.queue_func("transaction", tx.to_dict())
262			return result
263
264	1		handler_func, kwargs = handler_context, {}
265	1		if get_apm_name() == "es":
266			handler_func = handler_context_apm
267			kwargs = dict(apm_client=ElasticAPM.get_client())
268
269	1		async def inner(*args):
270			"""Inner decorated with events attribute."""
271	1		return await handler_func(args, *kwargs)
272	1		inner.events = [event]
273	1		inner.events.extend(events)
274	1		return inner
275
276	1		return decorator
277
278
279	1		def now(tzone=timezone.utc):
280			"""Return the current datetime (default to UTC).
281
282			Args:
283			tzone (datetime.timezone): Specific time zone used in datetime.
284
285			Returns:
286			datetime.datetime.now: Date time with specific time zone.
287
288			"""
289	1		return datetime.now(tzone)
290
291
292	1		def run_on_thread(method):
293			"""Decorate to run the decorated method inside a new thread.
294
295			Args:
296			method (function): function used to run as a new thread.
297
298			Returns:
299			Decorated method that will run inside a new thread.
300			When the decorated method is called, it will not return the created
301			thread to the caller.
302
303			"""
304	1		def threaded_method(*args):
305			"""Ensure the handler method runs inside a new thread."""
306	1		thread = Thread(target=method, args=args)
307
308			# Set daemon mode so that we don't have to wait for these threads
309			# to finish when exiting Kytos
310	1		thread.daemon = True
311	1		thread.start()
312	1		return threaded_method
313
314
315	1		def get_time(data=None):
316			"""Receive a dictionary or a string and return a datatime instance.
317
318			data = {"year": 2006,
319			"month": 11,
320			"day": 21,
321			"hour": 16,
322			"minute": 30 ,
323			"second": 00}
324
325			or
326
327			data = "21/11/06 16:30:00"
328
329			2018-04-17T17:13:50Z
330
331			Args:
332			data (str, dict): python dict or string to be converted to datetime
333
334			Returns:
335			datetime: datetime instance.
336
337			"""
338	1		if isinstance(data, str):
339	1		date = datetime.strptime(data, "%Y-%m-%dT%H:%M:%S")
340	1		elif isinstance(data, dict):
341	1		date = datetime(**data)
342			else:
343	1		return None
344	1		return date.replace(tzinfo=timezone.utc)
345
346
347	1		def _read_from_filename(yml_file_path: Path) -> dict:
348			"""Read from yml filename."""
349			spec_dict, _ = read_from_filename(yml_file_path)
350			return spec_dict
351
352
353	1		def load_spec(yml_file_path: Path):
354			"""Load and validate spec object given a yml file path."""
355	1		spec_dict = _read_from_filename(yml_file_path)
356	1		validate_spec(spec_dict)
357	1		return create_spec(spec_dict)
358
359
360	1		def validate_openapi(spec):
361			"""Decorator to validate a REST endpoint input.
362
363			Uses the schema defined in the openapi.yml file
364			to validate.
365			"""
366
367			def validate_decorator(func):
368			@functools.wraps(func)
369			def wrapper_validate(args, *kwargs):
370			try:
371			data = request.get_json()
372			except BadRequest:
373			result = "The request body is not a well-formed JSON."
374			raise BadRequest(result) from BadRequest
375			if data is None:
376			result = "The request body mimetype is not application/json."
377			raise UnsupportedMediaType(result)
378
379			openapi_request = FlaskOpenAPIRequest(request)
380			result = openapi_request_validator.validate(spec, openapi_request)
381			if result.errors:
382			error_response = (
383			"The request body contains invalid API data."
384			)
385			errors = result.errors[0]
386			if hasattr(errors, "schema_errors"):
387			schema_errors = errors.schema_errors[0]
388			error_log = {
389			"error_message": schema_errors.message,
390			"error_validator": schema_errors.validator,
391			"error_validator_value": schema_errors.validator_value,
392			"error_path": list(schema_errors.path),
393			"error_schema": schema_errors.schema,
394			"error_schema_path": list(schema_errors.schema_path),
395			}
396			LOG.debug(f"Invalid request (API schema): {error_log}")
397			error_response += f" {schema_errors.message} for field"
398			error_response += (
399			f" {'/'.join(map(str,schema_errors.path))}."
400			)
401			raise BadRequest(error_response) from BadRequest
402			return func(args, data=data, *kwargs)
403
404			return wrapper_validate
405
406			return validate_decorator
407

kytos-ng / kytos

Push — master ( ad4b11...eeb2a1 )

kytos.core.helpers.load_spec() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like