Python Utils

utils.py

import logging
import signal

logger = logging.getLogger(__name__)


class ExitSignalHandler(metaclass=Singleton):
    def __init__(self):
        self._cleanup_functions = []

        handled_signals = signal.SIGINT, signal.SIGTERM
        for handled_signal in handled_signals:
            self._attach_new_handler(handled_signal, self.run_cleanup_functions)

    def add_cleanup_function(self, function):
        self._cleanup_functions.append(function)

    def run_cleanup_functions(self, signum, frame):
        logger.info("Caught kill signal: {}".format(signum))
        for function in self._cleanup_functions:
            function()

    @staticmethod
    def _attach_new_handler(sig, new_handler):
        old_handler = signal.getsignal(sig) if callable(signal.getsignal(sig)) else None

        def handle_exit_signal(*args, **kwargs):
            new_handler(*args, **kwargs)
            if old_handler is not None:
                old_handler(*args, **kwargs)

        signal.signal(sig, handle_exit_signal)  # NOQA

import functools


def run_first(preceding_func):
    """ Runs some function before running decorated function """

    def wrapper_decorator(func):
        @functools.wraps(func)
        def decorator(*args, **kwargs):
            preceding_func()
            returned_value = func(*args, **kwargs)
            return returned_value

        return decorator

    return wrapper_decorator


def run_once(func):
    """ Runs the function only once (caches the return value for subsequent runs) """

    @functools.wraps(func)
    def decorator(*args, **kwargs):
        if decorator.has_run:
            return decorator.result
        decorator.result = func(*args, **kwargs)
        decorator.has_run = True
        return decorator.result

    decorator.has_run = False
    decorator.result = None
    return decorator


class Singleton(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
        return cls._instances[cls]