mercurial: comparison mercurial/ui.py

equal deleted inserted replaced

-:25fe689a4a64
+: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, **opts):
+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, **opts):
+def write_err(self, *args: bytes, **opts: _MsgOpts) -> None:
 self._write(self._ferr, *args, **opts)
-def _write(self, dest, *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, **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, **opts):
+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, **opts):
+def _writemsgnobuf(self, dest, *args: bytes, **opts: _MsgOpts) -> None:
 _writemsgwith(self._writenobuf, dest, *args, **opts)
-def flush(self):
+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=b' ', promptopts=None):
+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):
+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, default=0):
+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=None, default=None):
+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, **opts):
+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, **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, **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, **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, **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, label):
+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, stacklevel=1, config=None):
+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, version, stacklevel=2):
+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, **opts):
+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.
 """

changeset 49795	f1e820cda2f5
parent 49794	25fe689a4a64
child 49797	0449fb7729d7