rest_api_child

drunc.controller.children_interface.rest_api_child

Classes

AppCommander(app_name, app_host, app_port, response_host, response_port, proxy_host=None, proxy_port=None)

Source code in drunc/controller/children_interface/rest_api_child.py

def __init__(
    self,
    app_name: str,
    app_host: str,
    app_port: int,
    response_host: str,
    response_port: int,
    proxy_host: str = None,
    proxy_port: int = None,
):
    self.app_host = app_host
    self.app_port = app_port
    self.response_host = response_host
    self.response_port = response_port
    self.proxy_host = proxy_host
    self.proxy_port = proxy_port

    self.app = app_name
    self.log = get_logger(f"controller.{self.app}-commander")
    self.app_url = f"http://{self.app_host}:{self.app_port}/command"

    self.response_queue = queue.Queue()
    self.sent_cmd = None

Functions

check_response(timeout=0)

Check if a response is present in the queue

Parameters:

Name	Type	Description	Default
timeout	int	Timeout in seconds	0

Returns:

Name	Type	Description
dict	dict	Command response is json

Raises:

Type	Description
NoResponse	Description
ResponseTimeout	Description

Source code in drunc/controller/children_interface/rest_api_child.py

def check_response(self, timeout: int = 0) -> dict:
    """Check if a response is present in the queue

    Args:
        timeout (int, optional): Timeout in seconds

    Returns:
        dict: Command response is json

    Raises:
        NoResponse: Description
        ResponseTimeout: Description

    """
    try:
        # self.log.info(f"Checking for answers from {self.app} {self.sent_cmd}")
        r = self.response_queue.get(block=(timeout > 0), timeout=timeout)
        self.log.info(f"Received reply from {self.app} to {self.sent_cmd}")
        self.sent_cmd = None

    except queue.Empty:
        self.log.info(f"Queue empty! {self.app} to {self.sent_cmd}")
        if not timeout:
            raise NoResponse(
                f"No response available from {self.app} for command {self.sent_cmd}"
            )
        else:
            self.log.error(
                f"Timeout while waiting for a reply from {self.app} for command {self.sent_cmd}"
            )
            raise ResponseTimeout(
                f"Timeout while waiting for a reply from {self.app} for command {self.sent_cmd}"
            )
    return r

ResponseListener()

Source code in drunc/controller/children_interface/rest_api_child.py

def __init__(self):
    raise DruncSetupException("Call get() instead")

Functions

register(app, handler) classmethod

Register a new notification handler

:param app: The application :type app: str :param handler: The handler :type handler: { type_description }

:rtype: None

:raises RuntimeError: { exception_description }

Source code in drunc/controller/children_interface/rest_api_child.py

@classmethod
def register(cls, app: str, handler):
    """Register a new notification handler

    :param      app:           The application
    :type       app:           str
    :param      handler:       The handler
    :type       handler:       { type_description }

    :rtype:     None

    :raises     RuntimeError:  { exception_description }
    """
    if app in cls.handlers:
        raise DruncSetupException(
            f"Handler already registered with notification listerner for app {app}"
        )

    cls.handlers[app] = handler

unregister(app) classmethod

De-register a notification handler

Parameters:

Name	Type	Description	Default
app	str	application name	required

Source code in drunc/controller/children_interface/rest_api_child.py

@classmethod
def unregister(cls, app: str):
    """De-register a notification handler

    Args:
        app (str): application name

    """
    if app not in cls.handlers:
        raise DruncException(f"No handler registered for app {app}")
    del cls.handlers[app]

Functions