Source code for runway.cfngin.ui

"""CFNgin UI manipulation."""

from __future__ import annotations

import logging
import threading
from getpass import getpass
from typing import TYPE_CHECKING, Any, ContextManager, Optional, TextIO, Type, Union

if TYPE_CHECKING:
    from types import TracebackType

LOGGER = logging.getLogger(__name__)


[docs]def get_raw_input(message: str) -> str: """Just a wrapper for :func:`input` for testing purposes.""" return input(message)
[docs]class UI(ContextManager["UI"]): """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). """
[docs] def __init__(self) -> None: """Instantiate class.""" self._lock = threading.RLock()
[docs] def log( self, lvl: int, msg: Union[Exception, str], *args: Any, logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = LOGGER, **kwargs: Any, ) -> None: """Log the message if the current thread owns the underlying lock. Args: lvl: Log level. msg: String template or exception to use for the log record. logger: Specific logger to log to. """ with self: return logger.log(lvl, msg, *args, **kwargs)
[docs] def info( self, msg: str, *args: Any, logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = LOGGER, **kwargs: Any, ) -> None: """Log the line if the current thread owns the underlying lock. Args: msg: String template or exception to use for the log record. logger: Specific logger to log to. """ kwargs["logger"] = logger self.log(logging.INFO, msg, *args, **kwargs)
[docs] def ask(self, message: str) -> str: """Collect input from a user in a multithreaded environment. This wraps the built-in 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. """ with self: return get_raw_input(message)
[docs] def getpass(self, prompt: str, stream: Optional[TextIO] = None) -> str: """Wrap getpass to lock the UI.""" with self: return getpass(prompt, stream)
[docs] def __enter__(self) -> UI: """Enter the context manager.""" self._lock.__enter__() return self
[docs] def __exit__( self, exc_type: Optional[Type[BaseException]], exc_value: Optional[BaseException], traceback: Optional[TracebackType], ) -> None: """Exit the context manager.""" self._lock.__exit__(exc_type, exc_value, traceback)
# Global UI object for other modules to use. ui = UI()