Source code for runway.cfngin.ui

"""CFNgin UI manipulation."""
import logging
import threading
from getpass import getpass

from six.moves import input

LOGGER = logging.getLogger(__name__)


[docs]def get_raw_input(message): """Just a wrapper for :func:`input` for testing purposes.""" return input(message)
[docs]class UI(object): """Used internally from terminal output in a multithreaded environment. Ensures that two threads don't write over each other while asking a user for input (e.g. in interactive mode). """ def __init__(self): """Instantiate class.""" self._lock = threading.RLock()
[docs] def lock(self, *_args, **_kwargs): """Obtain an exclusive lock on the UI for the current thread.""" return self._lock.acquire()
# TODO uncomment signature when dropping python 2 # def log(self, lvl, msg, *args, logger=LOGGER, **kwargs):
[docs] def log(self, lvl, msg, *args, **kwargs): """Log the message if the current thread owns the underlying lock. Args: lvl (int): Log level. msg (Union[str, Exception]): String template or exception to use for the log record. Keyword Args: logger (Union[logging.LoggerAdaptor, logging.Logger]): Specific logger to log to. """ self.lock() logger = kwargs.pop("logger", LOGGER) try: return logger.log(lvl, msg, *args, **kwargs) finally: self.unlock()
[docs] def unlock(self, *_args, **_kwargs): """Release the lock on the UI.""" return self._lock.release()
# TODO uncomment signature when dropping python 2 # def info(self, msg, *args, logger=LOGGER, **kwargs):
[docs] def info(self, msg, *args, **kwargs): """Log the line if the current thread owns the underlying lock. Args: msg (Union[str, Exception]): String template or exception to use for the log record. Keyword Args: logger (Union[logging.LoggerAdaptor, logging.Logger]): Specific logger to log to. """ self.log(logging.INFO, msg, *args, **kwargs)
[docs] def ask(self, message): """Collect input from a user in a multithreaded environment. This wraps the built-in raw_input function to ensure that only 1 thread is asking for input from the user at a give time. Any process that tries to log output to the terminal will be blocked while the user is being prompted. """ self.lock() try: return get_raw_input(message) finally: self.unlock()
[docs] def getpass(self, *args): """Wrap getpass to lock the UI.""" try: self.lock() return getpass(*args) finally: self.unlock()
# Global UI object for other modules to use. ui = UI()