Util

Utility functions for odin-control.

This module implements utility methods for odin-control.

decode_request_body(request)

Decode the body of an HTTP request object.

This function decodes the body of the provided request object. If the request headers indicate that the request is of type "application/json", the body will be decoded from JSON. Otherwise, the body will be returned as-is.

Parameters:

Name	Type	Description	Default
request		HTTP request object	required

Returns:

Type	Description
	Decoded request body, either as a Python object if JSON or raw bytes otherwise

Source code in src/odin_control/util.py

def decode_request_body(request):
    """Decode the body of an HTTP request object.

    This function decodes the body of the provided request object. If the request headers indicate
    that the request is of type "application/json", the body will be decoded from JSON. Otherwise,
    the body will be returned as-is.

    :param request: HTTP request object
    :return: Decoded request body, either as a Python object if JSON or raw bytes otherwise
    """
    try:
        body_type = request.headers["Content-Type"]
        if body_type == "application/json":
            body = json_decode(request.body)
        else:
            body = request.body
    except (TypeError):
        body = request.body
    return body

get_async_event_loop()

Get the asyncio event loop.

This function obtains and returns the current asyncio event loop. If no loop is present, a new one is created and set as the event loop.

Returns:

Type	Description
	an asyncio event loop

Source code in src/odin_control/util.py

def get_async_event_loop():
    """Get the asyncio event loop.

    This function obtains and returns the current asyncio event loop. If no loop is present, a new
    one is created and set as the event loop.

    :return: an asyncio event loop
    """
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

    return loop

run_async(func, *args, **kwargs)

Run an async function synchronously in an event loop.

This function can be used to run an async function synchronously, i.e. without the need for an await() call. The function is run on an asyncio event loop and the result is returned.

Parameters:

Name	Type	Description	Default
func		async function to run	required
args		positional arguments to function	()
kwargs		: keyword arguments to function	{}

Returns:

Type	Description
	result of function

Source code in src/odin_control/util.py

def run_async(func, *args, **kwargs):
    """Run an async function synchronously in an event loop.

    This function can be used to run an async function synchronously, i.e. without the need for an
    await() call. The function is run on an asyncio event loop and the result is returned.

    :param func: async function to run
    :param args: positional arguments to function
    :param kwargs:: keyword arguments to function
    :return: result of function
    """
    loop = get_async_event_loop()
    result = loop.run_until_complete(func(*args, **kwargs))
    return result

run_in_executor(executor, func, *args)

Run a function asynchronously in an executor.

This method extends the behaviour of Tornado IOLoop equivalent to allow nested task execution without having to modify the underlying asyncio loop creation policy on python 3. If the current execution context does not have a valid IO loop, a new one will be created and used. The method returns a tornado Future instance, allowing it to be awaited in an async method where applicable.

Parameters:

Name	Type	Description	Default
executor		a concurrent.futures.Executor instance to run the task in	required
func		the function to execute	required
args		list of arguments to pass to the function	()

Returns:

Type	Description
	a Future wrapping the task

Source code in src/odin_control/util.py

def run_in_executor(executor, func, *args):
    """Run a function asynchronously in an executor.

    This method extends the behaviour of Tornado IOLoop equivalent to allow nested task execution
    without having to modify the underlying asyncio loop creation policy on python 3. If the
    current execution context does not have a valid IO loop, a new one will be created and used.
    The method returns a tornado Future instance, allowing it to be awaited in an async method where
    applicable.

    :param executor: a concurrent.futures.Executor instance to run the task in
    :param func: the function to execute
    :param args: list of arguments to pass to the function

    :return: a Future wrapping the task
    """
    # Try to get the current asyncio event loop, otherwise create a new one
    get_async_event_loop()

    # Run the function in the specified executor, handling tornado version 4 where there was no
    # run_in_executor implementation
    future = IOLoop.current().run_in_executor(executor, func, *args)

    return future