Source code for userinterfaces.buffer_interface

"""Non-interactive interface that stores output.

.. versionadded:: 6.4
"""
#
# (C) Pywikibot team, 2021-2024
#
# Distributed under the terms of the MIT license.
#
from __future__ import annotations

import logging
import queue
from typing import Any, Sequence

from pywikibot import config
from pywikibot.logging import INFO, VERBOSE
from pywikibot.userinterfaces._interface_base import ABUIC




class UI(ABUIC):

    """Collects output into an unseen buffer.

    .. versionadded:: 6.4
    """

    def __init__(self) -> None:
        """Initialize the UI."""
        super().__init__()

        self._buffer = queue.Queue()

        self.log_handler = logging.handlers.QueueHandler(self._buffer)
        self.log_handler.setLevel(VERBOSE if config.verbose_output else INFO)



    def flush(self) -> None:
        """Flush cached output."""
        self.clear()




    def init_handlers(self, root_logger, *args, **kwargs) -> None:
        """Initialize the handlers for user output."""
        root_logger.addHandler(self.log_handler)




    def input(self, question: str, password: bool = False,
              default: str = '', force: bool = False) -> str:
        """Ask the user a question and return the answer."""
        return default




    def input_choice(self, question: str, options, default: str | None = None,
                     return_shortcut: bool = True,
                     automatic_quit: bool = True, force: bool = False):
        """Ask the user and returns a value from the options."""
        return default




    def input_list_choice(self, question: str, answers: Sequence[Any],
                          default: int | str | None = None,
                          force: bool = False) -> Any:
        """Ask the user to select one entry from a list of entries."""
        return default




    def output(self, text, *args, **kwargs) -> None:
        """Output text that would usually go to a stream."""
        self._buffer.put(text)




    def pop_output(self):
        """Provide and clear any buffered output."""
        output = []

        while not self._buffer.empty():
            record = self._buffer.get_nowait()

            if isinstance(record, str):
                output.append(record)
            elif isinstance(record, logging.LogRecord):
                output.append(record.getMessage())
            else:
                raise ValueError('Buffer can only contain logs and strings, '
                                 f'had {type(record).__name__}')

        return output




    def clear(self) -> None:
        """Removes any buffered output."""
        self.pop_output()