comparison mercurial/ui.py @ 49795:f1e820cda2f5

typing: add type hints related to message output in mercurial/ui.py

This will shake loose some bytes vs str issues in the doc checker.

49794:25fe689a4a64

import subprocess
import sys
import traceback

from typing import (
    Optional,
)

from .i18n import _
from .node import hex
from .pycompat import (

    procutil,
    resourceutil,
    stringutil,
    urlutil,
)

urlreq = util.urlreq

# for use with str.translate(None, _keepalnum), to keep just alphanumerics
_keepalnum = b''.join(

    def canbatchlabeledwrites(self):
        '''check if write calls with labels are batchable'''
        # Windows color printing is special, see ``write``.
        return self._colormode != b'win32'

    def write(self, *args, **opts):
        """write args to output

        By default, this method simply writes to the buffer or stdout.
        Color mode can be set on the UI class to have the output decorated
        with color modifier before being written to stdout.

        finally:
            self._blockedtimes[b'stdio_blocked'] += (
                util.timer() - starttime
            ) * 1000

    def write_err(self, *args, **opts):
        self._write(self._ferr, *args, **opts)

    def _write(self, dest, *args, **opts):
        # update write() as well if you touch this code
        if self._isbuffered(dest):
            label = opts.get('label', b'')
            if label and self._bufferapplylabels:
                self._buffers[-1].extend(self.label(a, label) for a in args)
            else:
                self._buffers[-1].extend(args)
        else:
            self._writenobuf(dest, *args, **opts)

    def _writenobuf(self, dest, *args, **opts):
        # update write() as well if you touch this code
        if not opts.get('keepprogressbar', False):
            self._progclear()
        msg = b''.join(args)

        finally:
            self._blockedtimes[b'stdio_blocked'] += (
                util.timer() - starttime
            ) * 1000

    def _writemsg(self, dest, *args, **opts):
        timestamp = self.showtimestamp and opts.get('type') in {
            b'debug',
            b'error',
            b'note',
            b'status',

            ) + args
        _writemsgwith(self._write, dest, *args, **opts)
        if timestamp:
            dest.flush()

    def _writemsgnobuf(self, dest, *args, **opts):
        _writemsgwith(self._writenobuf, dest, *args, **opts)

    def flush(self):
        # opencode timeblockedsection because this is a critical path
        starttime = util.timer()
        try:
            try:
                self._fout.flush()

            # usually those are non-interactive
            return self._isatty(self._fout)

        return i

    def _readline(self, prompt=b' ', promptopts=None):
        # Replacing stdin/stdout temporarily is a hard problem on Python 3
        # because they have to be text streams with *no buffering*. Instead,
        # we use rawinput() only if call_readline() will be invoked by
        # PyOS_Readline(), so no I/O will be made at Python layer.
        usereadline = (

            return r
        except EOFError:
            raise error.ResponseExpected()

    @staticmethod
    def extractchoices(prompt):
        """Extract prompt message and list of choices from specified prompt.

        This returns tuple "(message, choices)", and "choices" is the
        list of tuple "(response character, text without &)".

            ampidx = s.index(b'&')
            return s[ampidx + 1 : ampidx + 2].lower(), s.replace(b'&', b'', 1)

        return (msg, [choicetuple(s) for s in choices])

    def promptchoice(self, prompt, default=0):
        """Prompt user with a message, read response, and ensure it matches
        one of the provided choices. The prompt is formatted as follows:

           "would you like fries with that (Yn)? $$ &Yes $$ &No"

            if r.lower() in resps:
                return resps.index(r.lower())
            # TODO: shouldn't it be a warning?
            self._writemsg(self._fmsgout, _(b"unrecognized response\n"))

    def getpass(self, prompt=None, default=None):
        if not self.interactive():
            return default
        try:
            self._writemsg(
                self._fmsgerr,

                else:
                    return util.get_password()
        except EOFError:
            raise error.ResponseExpected()

    def status(self, *msg, **opts):
        """write status message to output (if ui.quiet is False)

        This adds an output label of "ui.status".
        """
        if not self.quiet:
            self._writemsg(self._fmsgout, type=b'status', *msg, **opts)

    def warn(self, *msg, **opts):
        """write warning message to output (stderr)

        This adds an output label of "ui.warning".
        """
        self._writemsg(self._fmsgerr, type=b'warning', *msg, **opts)

    def error(self, *msg, **opts):
        """write error message to output (stderr)

        This adds an output label of "ui.error".
        """
        self._writemsg(self._fmsgerr, type=b'error', *msg, **opts)

    def note(self, *msg, **opts):
        """write note to output (if ui.verbose is True)

        This adds an output label of "ui.note".
        """
        if self.verbose:
            self._writemsg(self._fmsgout, type=b'note', *msg, **opts)

    def debug(self, *msg, **opts):
        """write debug message to output (if ui.debugflag is True)

        This adds an output label of "ui.debug".
        """
        if self.debugflag:

            for logger in activeloggers:
                logger.log(self, event, msg, opts)
        finally:
            self._loggers = registeredloggers

    def label(self, msg, label):
        """style msg based on supplied label

        If some color mode is enabled, this will add the necessary control
        characters to apply such color. In addition, 'debug' color mode adds
        markup showing which label affects a piece of text.

        """
        if self._colormode is not None:
            return color.colorlabel(self, msg, label)
        return msg

    def develwarn(self, msg, stacklevel=1, config=None):
        """issue a developer warning message

        Use 'stacklevel' to report the offender some layers further up in the
        stack.
        """

            # avoid cycles
            del curframe
            del calframe

    def deprecwarn(self, msg, version, stacklevel=2):
        """issue a deprecation warning

        - msg: message explaining what is deprecated and how to upgrade,
        - version: last version where the API will be supported,
        """

    if name == b'stderr':
        return ui.ferr, ui.ferr
    raise error.Abort(b'invalid ui.message-output destination: %s' % name)


def _writemsgwith(write, dest, *args, **opts):
    """Write ui message with the given ui._write*() function

    The specified message type is translated to 'ui.<type>' label if the dest
    isn't a structured channel, so that the message will be colorized.
    """

49795:f1e820cda2f5

import subprocess
import sys
import traceback

from typing import (
    Dict,
    List,
    Optional,
    Tuple,
    Union,
    cast,
)

from .i18n import _
from .node import hex
from .pycompat import (

    procutil,
    resourceutil,
    stringutil,
    urlutil,
)

# The **opts args of the various write() methods can be basically anything, but
# there's no way to express it as "anything but str".  So type it to be the
# handful of known types that are used.
_MsgOpts = Union[bytes, bool, List["_PromptChoice"]]
_PromptChoice = Tuple[bytes, bytes]

urlreq = util.urlreq

# for use with str.translate(None, _keepalnum), to keep just alphanumerics
_keepalnum = b''.join(

    def canbatchlabeledwrites(self):
        '''check if write calls with labels are batchable'''
        # Windows color printing is special, see ``write``.
        return self._colormode != b'win32'

    def write(self, *args: bytes, **opts: _MsgOpts) -> None:
        """write args to output

        By default, this method simply writes to the buffer or stdout.
        Color mode can be set on the UI class to have the output decorated
        with color modifier before being written to stdout.

        finally:
            self._blockedtimes[b'stdio_blocked'] += (
                util.timer() - starttime
            ) * 1000

    def write_err(self, *args: bytes, **opts: _MsgOpts) -> None:
        self._write(self._ferr, *args, **opts)

    def _write(self, dest, *args: bytes, **opts: _MsgOpts) -> None:
        # update write() as well if you touch this code
        if self._isbuffered(dest):
            label = opts.get('label', b'')
            if label and self._bufferapplylabels:
                self._buffers[-1].extend(self.label(a, label) for a in args)
            else:
                self._buffers[-1].extend(args)
        else:
            self._writenobuf(dest, *args, **opts)

    def _writenobuf(self, dest, *args: bytes, **opts: _MsgOpts) -> None:
        # update write() as well if you touch this code
        if not opts.get('keepprogressbar', False):
            self._progclear()
        msg = b''.join(args)

        finally:
            self._blockedtimes[b'stdio_blocked'] += (
                util.timer() - starttime
            ) * 1000

    def _writemsg(self, dest, *args: bytes, **opts: _MsgOpts) -> None:
        timestamp = self.showtimestamp and opts.get('type') in {
            b'debug',
            b'error',
            b'note',
            b'status',

            ) + args
        _writemsgwith(self._write, dest, *args, **opts)
        if timestamp:
            dest.flush()

    def _writemsgnobuf(self, dest, *args: bytes, **opts: _MsgOpts) -> None:
        _writemsgwith(self._writenobuf, dest, *args, **opts)

    def flush(self) -> None:
        # opencode timeblockedsection because this is a critical path
        starttime = util.timer()
        try:
            try:
                self._fout.flush()

            # usually those are non-interactive
            return self._isatty(self._fout)

        return i

    def _readline(
        self,
        prompt: bytes = b' ',
        promptopts: Optional[Dict[str, _MsgOpts]] = None,
    ) -> bytes:
        # Replacing stdin/stdout temporarily is a hard problem on Python 3
        # because they have to be text streams with *no buffering*. Instead,
        # we use rawinput() only if call_readline() will be invoked by
        # PyOS_Readline(), so no I/O will be made at Python layer.
        usereadline = (

            return r
        except EOFError:
            raise error.ResponseExpected()

    @staticmethod
    def extractchoices(prompt: bytes) -> Tuple[bytes, List[_PromptChoice]]:
        """Extract prompt message and list of choices from specified prompt.

        This returns tuple "(message, choices)", and "choices" is the
        list of tuple "(response character, text without &)".

            ampidx = s.index(b'&')
            return s[ampidx + 1 : ampidx + 2].lower(), s.replace(b'&', b'', 1)

        return (msg, [choicetuple(s) for s in choices])

    def promptchoice(self, prompt: bytes, default: int = 0) -> int:
        """Prompt user with a message, read response, and ensure it matches
        one of the provided choices. The prompt is formatted as follows:

           "would you like fries with that (Yn)? $$ &Yes $$ &No"

            if r.lower() in resps:
                return resps.index(r.lower())
            # TODO: shouldn't it be a warning?
            self._writemsg(self._fmsgout, _(b"unrecognized response\n"))

    def getpass(
        self, prompt: Optional[bytes] = None, default: Optional[bytes] = None
    ) -> Optional[bytes]:
        if not self.interactive():
            return default
        try:
            self._writemsg(
                self._fmsgerr,

                else:
                    return util.get_password()
        except EOFError:
            raise error.ResponseExpected()

    def status(self, *msg: bytes, **opts: _MsgOpts) -> None:
        """write status message to output (if ui.quiet is False)

        This adds an output label of "ui.status".
        """
        if not self.quiet:
            self._writemsg(self._fmsgout, type=b'status', *msg, **opts)

    def warn(self, *msg: bytes, **opts: _MsgOpts) -> None:
        """write warning message to output (stderr)

        This adds an output label of "ui.warning".
        """
        self._writemsg(self._fmsgerr, type=b'warning', *msg, **opts)

    def error(self, *msg: bytes, **opts: _MsgOpts) -> None:
        """write error message to output (stderr)

        This adds an output label of "ui.error".
        """
        self._writemsg(self._fmsgerr, type=b'error', *msg, **opts)

    def note(self, *msg: bytes, **opts: _MsgOpts) -> None:
        """write note to output (if ui.verbose is True)

        This adds an output label of "ui.note".
        """
        if self.verbose:
            self._writemsg(self._fmsgout, type=b'note', *msg, **opts)

    def debug(self, *msg: bytes, **opts: _MsgOpts) -> None:
        """write debug message to output (if ui.debugflag is True)

        This adds an output label of "ui.debug".
        """
        if self.debugflag:

            for logger in activeloggers:
                logger.log(self, event, msg, opts)
        finally:
            self._loggers = registeredloggers

    def label(self, msg: bytes, label: bytes) -> bytes:
        """style msg based on supplied label

        If some color mode is enabled, this will add the necessary control
        characters to apply such color. In addition, 'debug' color mode adds
        markup showing which label affects a piece of text.

        """
        if self._colormode is not None:
            return color.colorlabel(self, msg, label)
        return msg

    def develwarn(
        self, msg: bytes, stacklevel: int = 1, config: Optional[bytes] = None
    ) -> None:
        """issue a developer warning message

        Use 'stacklevel' to report the offender some layers further up in the
        stack.
        """

            # avoid cycles
            del curframe
            del calframe

    def deprecwarn(
        self, msg: bytes, version: bytes, stacklevel: int = 2
    ) -> None:
        """issue a deprecation warning

        - msg: message explaining what is deprecated and how to upgrade,
        - version: last version where the API will be supported,
        """

    if name == b'stderr':
        return ui.ferr, ui.ferr
    raise error.Abort(b'invalid ui.message-output destination: %s' % name)


def _writemsgwith(write, dest, *args: bytes, **opts: _MsgOpts) -> None:
    """Write ui message with the given ui._write*() function

    The specified message type is translated to 'ui.<type>' label if the dest
    isn't a structured channel, so that the message will be colorized.
    """