kytos.core.helpers.avalidate_openapi_request() - Code Metrics - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

kytos.core.helpers.avalidate_openapi_request() A
last analyzed 2025-06-18 19:57 UTC

↳ Parent: kytos.core.helpers

Complexity

Conditions

Size

Total Lines	25
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	12
nop	2
dl	0
loc	25
rs	9.8
c	0
b	0
f	0

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

from openapi_core import Spec, unmarshal_request
from openapi_core.exceptions import OpenAPIError
from openapi_spec_validator import validate_spec
from openapi_spec_validator.readers import read_from_filename

from kytos.core.apm import ElasticAPM
from kytos.core.config import KytosConfig
from kytos.core.rest_api import (AStarletteOpenAPIRequest, HTTPException,
                                 Request, StarletteOpenAPIRequest,
                                 content_type_json_or_415, get_body)

__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()}

ds_executors = defaultdict(lambda: ThreadPoolExecutor(max_workers=1,
                           thread_name_prefix="dynamic_single"))


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()
                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()
                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 = {"apm_client": ElasticAPM.get_client()}

        def get_executor(pool, event, default_pool="app", handler=handler):
            """Get executor."""
            if pool == "dynamic_single":
                return ds_executors[handler]
            if pool and pool in executors:
                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"]
            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()
                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()
                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 = {"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 = _read_from_filename(yml_file_path)
    validate_spec(spec)
    return Spec.from_dict(spec)


def _request_validation_result_or_400(errors: OpenAPIError) -> None:
    """Raise HTTP 400."""
    error_response = "The request body contains invalid API data."
    if not errors.__cause__:
        error_response = str(errors)
    elif (hasattr(errors.__cause__, "schema_errors") and
            errors.__cause__.schema_errors):
        schema_errors = errors.__cause__.schema_errors
        for error in schema_errors:
            error_response += f", {error.message} for field"
            error_response += f" {'/'.join(map(str,error.path))}."
    raise HTTPException(400, detail=error_response)


def validate_openapi_request(
    spec: Spec, request: Request, loop: AbstractEventLoop
) -> bytes:
    """Validate a Request given an OpenAPI spec.

    This function is meant to be called from a synchronous context
    since StarletteOpenAPIRequest internally uses `asgiref.sync.AsyncToSync`
    and its forcing not to use the current running event loop.
    """
    body = get_body(request, loop)
    if body:
        content_type_json_or_415(request)
    openapi_request = StarletteOpenAPIRequest(request, body)
    try:
        unmarshal_request(openapi_request, spec)
    except OpenAPIError as err:
        _request_validation_result_or_400(err)
    return body


async def avalidate_openapi_request(
        spec: Spec,
        request: Request,
) -> bytes:
    """Async validate_openapi_request.

    This function is for async routes. It also returns the request body bytes.
    It does not try to assume that it'll have a loadable json body to work
    seamlessly with as many type of endpoints with minimal friction.
    You can use await aget_json_or_400(request) to get the request body.

    Example:

    await avalidate_openapi_request(self.spec, request)
    body = await aget_json_or_400(request)
    """
    body = await request.body()
    if body:
        content_type_json_or_415(request)
    openapi_request = AStarletteOpenAPIRequest(request, body)
    try:
        unmarshal_request(openapi_request, spec)
    except OpenAPIError as err:
        _request_validation_result_or_400(err)
    return body


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):
            request: Request = None
            napp = None
            for arg in args:
                if isinstance(arg, Request):
                    request = arg
                if hasattr(arg, "controller"):
                    napp = arg
                if request and napp:
                    break
            if not request:
                err = f"{func.__name__} args doesn't have a Request argument"
                raise RuntimeError(err)
            if not napp:
                err = f"{func.__name__} should be a NApp method to get ev_loop"
                raise RuntimeError(err)
            validate_openapi_request(spec, request, napp.controller.loop)
            return func(*args, **kwargs)
        return wrapper_validate
    return validate_decorator


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

kytos-ng / kytos

kytos.core.helpers.avalidate_openapi_request() A last analyzed 2025-06-18 19:57 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

kytos.core.helpers.avalidate_openapi_request() A
last analyzed 2025-06-18 19:57 UTC