upstream/mercurial-mirror Commit - r52180:f15cb511

pytype: move some type comment to proper annotation...

marmoute -

r52180:f15cb511 default

parent child

mercurial/branchmap.py

0 +10 -9

@@ -196,15 +196,16 b' class branchcache:'
196		196
197	def __init__(	197	def __init__(
198	self,	198	self,
199	repo,	199	repo: "localrepo.localrepository",
200	entries=~~(),~~	200	entries: Union[
201	tipnode=None,	201	Dict[bytes, List[bytes]], Iterable[Tuple[bytes, List[bytes]]]
202	tiprev=nullrev,	202	] = (),
203	filteredhash=None,	203	tipnode: Optional[bytes] = None,
204	closednodes=None,	204	tiprev: Optional[int] = nullrev,
205	hasnode=None,	205	filteredhash: Optional[bytes] = None,
206	):	206	closednodes: Optional[Set[bytes]] = None,
207	# type: (localrepo.localrepository, Union[Dict[bytes, List[bytes]], Iterable[Tuple[bytes, List[bytes]]]], bytes, int, Optional[bytes], Optional[Set[bytes]], Optional[Callable[[bytes], bool]]) -> None	207	hasnode: Optional[Callable[[bytes], bool]] = None,
		208	) -> None:
208	"""hasnode is a function which can be used to verify whether changelog	209	"""hasnode is a function which can be used to verify whether changelog
209	has a given node or not. If it's not provided, we assume that every node	210	has a given node or not. If it's not provided, we assume that every node
210	we have exists in changelog"""	211	we have exists in changelog"""

mercurial/cmdutil.py

0 +4 -2

                 return 0
-            def readgraftstate(repo, graftstate):
+            def readgraftstate(
-                # type: (Any, statemod.cmdstate) -> Dict[bytes, Any]
+                repo: Any,
+                graftstate: statemod.cmdstate,
+            ) -> Dict[bytes, Any]:
                 """read the graft state file and return a dict of the data stored in it"""
                 try:
                     return graftstate.read()

mercurial/encoding.py

0 +23 -36

             assert all(i.startswith((b"\xe2", b"\xef")) for i in _ignore)
-            def hfsignoreclean(s):
+            def hfsignoreclean(s: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 """Remove codepoints ignored by HFS+ from s.
                 >>> hfsignoreclean(u'.h\u200cg'.encode('utf-8'))
                 if typing.TYPE_CHECKING:
                     # pseudo implementation to help pytype see localstr() constructor
-                    def __init__(self, u, l):
+                    def __init__(self, u: bytes, l: bytes) -> None:
-                        # type: (bytes, bytes) -> None
                         super(localstr, self).__init__(l)
                         self._utf8 = u
                 """
-            def tolocal(s):
+            def tolocal(s: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 """
                 Convert a string from internal UTF-8 to local encoding
                     )
-            def fromlocal(s):
+            def fromlocal(s: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 """
                 Convert a string from the local character encoding to UTF-8
                     )
-            def unitolocal(u):
+            def unitolocal(u: str) -> bytes:
-                # type: (Text) -> bytes
                 """Convert a unicode string to a byte string of local encoding"""
                 return tolocal(u.encode('utf-8'))
-            def unifromlocal(s):
+            def unifromlocal(s: bytes) -> str:
-                # type: (bytes) -> Text
                 """Convert a byte string of local encoding to a unicode string"""
                 return fromlocal(s).decode('utf-8')
-            def unimethod(bytesfunc):
+            def unimethod(bytesfunc: Callable[[Any], bytes]) -> Callable[[Any], str]:
-                # type: (Callable[[Any], bytes]) -> Callable[[Any], Text]
                 """Create a proxy method that forwards __unicode__() and __str__() of
                 Python 3 to __bytes__()"""
             strmethod = unimethod
-            def lower(s):
+            def lower(s: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 """best-effort encoding-aware case-folding of local string s"""
                 try:
                     return asciilower(s)
                     )
-            def upper(s):
+            def upper(s: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 """best-effort encoding-aware case-folding of local string s"""
                 try:
                     return asciiupper(s)
                     return upperfallback(s)
-            def upperfallback(s):
+            def upperfallback(s: Any) -> Any:
-                # type: (Any) -> Any
                 try:
                     if isinstance(s, localstr):
                         u = s._utf8.decode("utf-8")
             )
-            def colwidth(s):
+            def colwidth(s: bytes) -> int:
-                # type: (bytes) -> int
                 """Find the column width of a string for display in the local encoding"""
                 return ucolwidth(s.decode(_sysstr(encoding), 'replace'))
-            def ucolwidth(d):
+            def ucolwidth(d: Text) -> int:
-                # type: (Text) -> int
                 """Find the column width of a Unicode string for display"""
                 eaw = getattr(unicodedata, 'east_asian_width', None)
                 if eaw is not None:
                 return len(d)
-            def getcols(s, start, c):
+            def getcols(s: bytes, start: int, c: int) -> bytes:
-                # type: (bytes, int, int) -> bytes
                 """Use colwidth to find a c-column substring of s starting at byte
                 index start"""
                 for x in range(start + c, len(s)):
                 raise ValueError('substring not found')
-            def trim(s, width, ellipsis=b'', leftside=False):
+            def trim(
-                # type: (bytes, int, bytes, bool) -> bytes
+                s: bytes,
+                width: int,
+                ellipsis: bytes = b'',
+                leftside: bool = False,
+            ) -> bytes:
                 """Trim string 's' to at most 'width' columns (including 'ellipsis').
                 If 'leftside' is True, left side of string 's' is trimmed.
                 other = 0
-            def jsonescape(s, paranoid=False):
+            def jsonescape(s: Any, paranoid: Any = False) -> Any:
-                # type: (Any, Any) -> Any
                 """returns a string suitable for JSON
                 JSON is problematic for us because it doesn't support non-Unicode
             _utf8len = [0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 2, 2, 3, 4]
-            def getutf8char(s, pos):
+            def getutf8char(s: bytes, pos: int) -> bytes:
-                # type: (bytes, int) -> bytes
                 """get the next full utf-8 character in the given string, starting at pos
                 Raises a UnicodeError if the given location does not start a valid
                 return c
-            def toutf8b(s):
+            def toutf8b(s: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 """convert a local, possibly-binary string into UTF-8b
                 This is intended as a generic method to preserve data when working
                 return bytes(r)
-            def fromutf8b(s):
+            def fromutf8b(s: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 """Given a UTF-8b string, return a local, possibly-binary string.
                 return the original binary string. This

mercurial/error.py

0 +68 -55

             ]
-            def _tobytes(exc):
+            def _tobytes(exc) -> bytes:
-                # type: (...) -> bytes
                 """Byte-stringify exception in the same way as BaseException_str()"""
                 if not exc.args:
                     return b''
                 coarse_exit_code = None
                 detailed_exit_code = None
-                def __init__(self, message, hint=None):
+                def __init__(self, message: bytes, hint: Optional[bytes] = None) -> None:
-                    # type: (bytes, Optional[bytes]) -> None
                     self.message = message
                     self.hint = hint
                     # Pass the message into the Exception constructor to help extensions
                 def __bytes__(self):
                     return self.message
-                def __str__(self):
+                def __str__(self) -> str:
-                    # type: () -> str
                     # the output would be unreadable if the message was translated,
                     # but do not replace it with encoding.strfromlocal(), which
                     # may raise another exception.
                     return pycompat.sysstr(self.__bytes__())
-                def format(self):
+                def format(self) -> bytes:
-                    # type: () -> bytes
                     from .i18n import _
                     message = _(b"abort: %s\n") % self.message
             class SidedataHashError(RevlogError):
-                def __init__(self, key, expected, got):
+                def __init__(self, key: int, expected: bytes, got: bytes) -> None:
-                    # type: (int, bytes, bytes) -> None
                     self.hint = None
                     self.sidedatakey = key
                     self.expecteddigest = expected
             class LookupError(RevlogError, KeyError):
-                def __init__(self, name, index, message):
+                def __init__(self, name: bytes, index: bytes, message: bytes) -> None:
-                    # type: (bytes, bytes, bytes) -> None
                     self.name = name
                     self.index = index
                     # this can't be called 'message' because at least some installs of
             class CommandError(Exception):
                 """Exception raised on errors in parsing the command line."""
-                def __init__(self, command, message):
+                def __init__(self, command: Optional[bytes], message: bytes) -> None:
-                    # type: (Optional[bytes], bytes) -> None
                     self.command = command
                     self.message = message
                     super(CommandError, self).__init__()
             class UnknownCommand(Exception):
                 """Exception raised if command is not in the command table."""
-                def __init__(self, command, all_commands=None):
+                def __init__(
-                    # type: (bytes, Optional[List[bytes]]) -> None
+                    self,
+                    command: bytes,
+                    all_commands: Optional[List[bytes]] = None,
+                ) -> None:
                     self.command = command
                     self.all_commands = all_commands
                     super(UnknownCommand, self).__init__()
             class AmbiguousCommand(Exception):
                 """Exception raised if command shortcut matches more than one command."""
-                def __init__(self, prefix, matches):
+                def __init__(self, prefix: bytes, matches: List[bytes]) -> None:
-                    # type: (bytes, List[bytes]) -> None
                     self.prefix = prefix
                     self.matches = matches
                     super(AmbiguousCommand, self).__init__()
             class WorkerError(Exception):
                 """Exception raised when a worker process dies."""
-                def __init__(self, status_code):
+                def __init__(self, status_code: int) -> None:
-                    # type: (int) -> None
                     self.status_code = status_code
                     # Pass status code to superclass just so it becomes part of __bytes__
                     super(WorkerError, self).__init__(status_code)
                 coarse_exit_code = 1
                 detailed_exit_code = 240
-                def format(self):
+                def format(self) -> bytes:
-                    # type: () -> bytes
                     from .i18n import _
                     message = _(b"%s\n") % self.message
             class ConflictResolutionRequired(InterventionRequired):
                 """Exception raised when a continuable command required merge conflict resolution."""
-                def __init__(self, opname):
+                def __init__(self, opname: bytes) -> None:
-                    # type: (bytes) -> None
                     from .i18n import _
                     self.opname = opname
                 detailed_exit_code = 30
-                def __init__(self, message, location=None, hint=None):
+                def __init__(
-                    # type: (bytes, Optional[bytes], Optional[bytes]) -> None
+                    self,
+                    message: bytes,
+                    location: Optional[bytes] = None,
+                    hint: Optional[bytes] = None,
+                ) -> None:
                     super(ConfigError, self).__init__(message, hint=hint)
                     self.location = location
-                def format(self):
+                def format(self) -> bytes:
-                    # type: () -> bytes
                     from .i18n import _
                     if self.location is not None:
             class OutOfBandError(RemoteError):
                 """Exception raised when a remote repo reports failure"""
-                def __init__(self, message=None, hint=None):
+                def __init__(
-                    # type: (Optional[bytes], Optional[bytes]) -> None
+                    self,
+                    message: Optional[bytes] = None,
+                    hint: Optional[bytes] = None,
+                ):
                     from .i18n import _
                     if message:
                 detailed_exit_code = 10
-                def __init__(self, message, location=None, hint=None):
+                def __init__(
-                    # type: (bytes, Optional[Union[bytes, int]], Optional[bytes]) -> None
+                    self,
+                    message: bytes,
+                    location: Optional[Union[bytes, int]] = None,
+                    hint: Optional[bytes] = None,
+                ):
                     super(ParseError, self).__init__(message, hint=hint)
                     self.location = location
-                def format(self):
+                def format(self) -> bytes:
-                    # type: () -> bytes
                     from .i18n import _
                     if self.location is not None:
                 __bytes__ = _tobytes
-            def getsimilar(symbols, value):
+            def getsimilar(symbols: Iterable[bytes], value: bytes) -> List[bytes]:
-                # type: (Iterable[bytes], bytes) -> List[bytes]
                 sim = lambda x: difflib.SequenceMatcher(None, value, x).ratio()
                 # The cutoff for similarity here is pretty arbitrary. It should
                 # probably be investigated and tweaked.
                 return [s for s in symbols if sim(s) > 0.6]
-            def similarity_hint(similar):
+            def similarity_hint(similar: List[bytes]) -> Optional[bytes]:
-                # type: (List[bytes]) -> Optional[bytes]
                 from .i18n import _
                 if len(similar) == 1:
             class UnknownIdentifier(ParseError):
                 """Exception raised when a {rev,file}set references an unknown identifier"""
-                def __init__(self, function, symbols):
+                def __init__(self, function: bytes, symbols: Iterable[bytes]) -> None:
-                    # type: (bytes, Iterable[bytes]) -> None
                     from .i18n import _
                     similar = getsimilar(symbols, function)
             class StdioError(IOError):
                 """Raised if I/O to stdout or stderr fails"""
-                def __init__(self, err):
+                def __init__(self, err: IOError) -> None:
-                    # type: (IOError) -> None
                     IOError.__init__(self, err.errno, err.strerror)
                 # no __bytes__() because error message is derived from the standard IOError
             class UnsupportedMergeRecords(Abort):
-                def __init__(self, recordtypes):
+                def __init__(self, recordtypes: Iterable[bytes]) -> None:
-                    # type: (Iterable[bytes]) -> None
                     from .i18n import _
                     self.recordtypes = sorted(recordtypes)
             class UnknownVersion(Abort):
                 """generic exception for aborting from an encounter with an unknown version"""
-                def __init__(self, msg, hint=None, version=None):
+                def __init__(
-                    # type: (bytes, Optional[bytes], Optional[bytes]) -> None
+                    self,
+                    msg: bytes,
+                    hint: Optional[bytes] = None,
+                    version: Optional[bytes] = None,
+                ) -> None:
                     self.version = version
                     super(UnknownVersion, self).__init__(msg, hint=hint)
             class LockError(IOError):
-                def __init__(self, errno, strerror, filename, desc):
+                def __init__(
-                    # _type: (int, str, bytes, bytes) -> None
+                    self,
+                    errno: int,
+                    strerror: str,
+                    filename: bytes,
+                    desc: Optional[bytes],
+                ) -> None:
                     IOError.__init__(self, errno, strerror, filename)
                     self.desc = desc
             class LockHeld(LockError):
-                def __init__(self, errno, filename, desc, locker):
+                def __init__(
+                    self,
+                    errno: int,
+                    filename: bytes,
+                    desc: Optional[bytes],
+                    locker,
+                ):
                     LockError.__init__(self, errno, 'Lock held', filename, desc)
+                    self.filename: bytes = filename
                     self.locker = locker
             class ProgrammingError(Hint, RuntimeError):
                 """Raised if a mercurial (core or extension) developer made a mistake"""
-                def __init__(self, msg, *args, **kwargs):
+                def __init__(self, msg: AnyStr, *args, **kwargs):
-                    # type: (AnyStr, Any, Any) -> None
                     # On Python 3, turn the message back into a string since this is
                     # an internal-only error that won't be printed except in a
                     # stack traces.
                 Also contains the tombstone data substituted for the uncensored data.
                 """
-                def __init__(self, filename, node, tombstone):
+                def __init__(self, filename: bytes, node: bytes, tombstone: bytes):
-                    # type: (bytes, bytes, bytes) -> None
                     from .node import short
                     StorageError.__init__(self, b'%s:%s' % (filename, short(node)))
                 The error is a formatter string and an optional iterable of arguments.
                 """
-                def __init__(self, message, args=None):
+                def __init__(
-                    # type: (bytes, Optional[Sequence[bytes]]) -> None
+                    self,
+                    message: bytes,
+                    args: Optional[Sequence[bytes]] = None,
+                ) -> None:
                     self.message = message
                     self.messageargs = args

mercurial/hgweb/webcommands.py

0 +1 -2

             rev = webcommand(b'rev')(changeset)
-            def decodepath(path):
+            def decodepath(path: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 """Hook for mapping a path in the repository to a path in the
                 working copy.

mercurial/i18n.py

0 +6 -3

             _msgcache = {}  # encoding: {message: translation}
-            def gettext(message):
+            def gettext(message: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 """Translate message.
                 The message is looked up in the catalog to get a Unicode string,
             if _plain():
-                _ = lambda message: message  # type: Callable[[bytes], bytes]
+                def _(message: bytes) -> bytes:
+                    return message
             else:
                 _ = gettext

mercurial/logcmdutil.py

0 +15 -6

                 limit = attr.ib(default=None)
-            def parseopts(ui, pats, opts):
+            def parseopts(
-                # type: (Any, Sequence[bytes], Dict[bytes, Any]) -> walkopts
+                ui: Any,
+                pats: Sequence[bytes],
+                opts: Dict[bytes, Any],
+            ) -> walkopts:
                 """Parse log command options into walkopts
                 The returned walkopts will be passed in to getrevs() or makewalker().
                 return revs
-            def makewalker(repo, wopts):
+            def makewalker(
-                # type: (Any, walkopts) -> Tuple[smartset.abstractsmartset, Optional[Callable[[Any], matchmod.basematcher]]]
+                repo: Any,
+                wopts: walkopts,
+            ) -> Tuple[
+                smartset.abstractsmartset, Optional[Callable[[Any], matchmod.basematcher]]
+            ]:
                 """Build (revs, makefilematcher) to scan revision/file history
                 - revs is the smartset to be traversed.
                 return revs, filematcher
-            def getrevs(repo, wopts):
+            def getrevs(
-                # type: (Any, walkopts) -> Tuple[smartset.abstractsmartset, Optional[changesetdiffer]]
+                repo: Any,
+                wopts: walkopts,
+            ) -> Tuple[smartset.abstractsmartset, Optional[changesetdiffer]]:
                 """Return (revs, differ) where revs is a smartset
                 differ is a changesetdiffer with pre-configured file matcher.

mercurial/mail.py

0 +43 -28

             from typing import (
                 Any,
                 List,
+                Optional,
                 Tuple,
                 Union,
             )
                     return new_socket
-            def _pyhastls():
+            def _pyhastls() -> bool:
-                # type: () -> bool
                 """Returns true iff Python has TLS support, false otherwise."""
                 try:
                     import ssl
                         )
-            def codec2iana(cs):
+            def codec2iana(cs: str) -> str:
-                # type: (str) -> str
                 ''' '''
                 cs = email.charset.Charset(cs).input_charset.lower()
                 return cs
-            def mimetextpatch(s, subtype='plain', display=False):
+            def mimetextpatch(
-                # type: (bytes, str, bool) -> email.message.Message
+                s: bytes,
+                subtype: str = 'plain',
+                display: bool = False,
+            ) -> email.message.Message:
                 """Return MIME message suitable for a patch.
                 Charset will be detected by first trying to decode as us-ascii, then utf-8,
                 and finally the global encodings. If all those fail, fall back to
                 return mimetextqp(s, subtype, "iso-8859-1")
-            def mimetextqp(body, subtype, charset):
+            def mimetextqp(
-                # type: (bytes, str, str) -> email.message.Message
+                body: bytes, subtype: str, charset: str
+            ) -> email.message.Message:
                 """Return MIME message.
                 Quoted-printable transfer encoding will be used if necessary.
                 """
                 return msg
-            def _charsets(ui):
+            def _charsets(ui: Any) -> List[str]:
-                # type: (Any) -> List[str]
                 '''Obtains charsets to send mail parts not containing patches.'''
                 charsets = [
                     pycompat.sysstr(cs.lower())
                 return [cs for cs in charsets if not cs.endswith('ascii')]
-            def _encode(ui, s, charsets):
+            def _encode(ui: Any, s: bytes, charsets: List[str]) -> Tuple[bytes, str]:
-                # type: (Any, bytes, List[str]) -> Tuple[bytes, str]
                 """Returns (converted) string, charset tuple.
                 Finds out best charset by cycling through sendcharsets in descending
                 order. Tries both encoding and fallbackencoding for input. Only as
                 return s, 'us-ascii'
-            def headencode(ui, s, charsets=None, display=False):
+            def headencode(
-                # type: (Any, Union[bytes, str], List[str], bool) -> str
+                ui: Any,
+                s: Union[bytes, str],
+                charsets: Optional[List[str]] = None,
+                display: bool = False,
+            ) -> str:
                 '''Returns RFC-2047 compliant header from given string.'''
                 if not display:
                     # split into words?
                 return encoding.strfromlocal(s)
-            def _addressencode(ui, name, addr, charsets=None):
+            def _addressencode(
-                # type: (Any, str, str, List[str]) -> str
+                ui: Any, name: str, addr: str, charsets: Optional[List[str]] = None
+            ) -> str:
                 addr = encoding.strtolocal(addr)
                 name = headencode(ui, name, charsets)
                 try:
                 return email.utils.formataddr((name, encoding.strfromlocal(addr)))
-            def addressencode(ui, address, charsets=None, display=False):
+            def addressencode(
-                # type: (Any, bytes, List[str], bool) -> str
+                ui: Any,
+                address: bytes,
+                charsets: Optional[List[str]] = None,
+                display: bool = False,
+            ) -> str:
                 '''Turns address into RFC-2047 compliant header.'''
                 if display or not address:
                     return encoding.strfromlocal(address or b'')
                 return _addressencode(ui, name, addr, charsets)
-            def addrlistencode(ui, addrs, charsets=None, display=False):
+            def addrlistencode(
-                # type: (Any, List[bytes], List[str], bool) -> List[str]
+                ui: Any,
+                addrs: List[bytes],
+                charsets: Optional[List[str]] = None,
+                display: bool = False,
+            ) -> List[str]:
                 """Turns a list of addresses into a list of RFC-2047 compliant headers.
                 A single element of input list may contain multiple addresses, but output
                 always has one address per item"""
                 return result
-            def mimeencode(ui, s, charsets=None, display=False):
+            def mimeencode(
-                # type: (Any, bytes, List[str], bool) -> email.message.Message
+                ui: Any,
+                s: bytes,
+                charsets: Optional[List[str]] = None,
+                display: bool = False,
+            ) -> email.message.Message:
                 """creates mime text object, encodes it if needed, and sets
                 charset and transfer-encoding accordingly."""
                 cs = 'us-ascii'
             Generator = email.generator.BytesGenerator
-            def parse(fp):
+            def parse(fp: Any) -> email.message.Message:
-                # type: (Any) -> email.message.Message
                 ep = email.parser.Parser()
                 # disable the "universal newlines" mode, which isn't binary safe.
                 # I have no idea if ascii/surrogateescape is correct, but that's
                     fp.detach()
-            def parsebytes(data):
+            def parsebytes(data: bytes) -> email.message.Message:
-                # type: (bytes) -> email.message.Message
                 ep = email.parser.BytesParser()
                 return ep.parsebytes(data)
-            def headdecode(s):
+            def headdecode(s: Union[email.header.Header, bytes]) -> bytes:
-                # type: (Union[email.header.Header, bytes]) -> bytes
                 '''Decodes RFC-2047 header'''
                 uparts = []
                 for part, charset in email.header.decode_header(s):

mercurial/pathutil.py

0 +14 -18

             ]
-            def _lowerclean(s):
+            def _lowerclean(s: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 return encoding.hfsignoreclean(s.lower())
                     else:
                         self.normcase = lambda x: x
-                def __call__(self, path, mode=None):
+                def __call__(self, path: bytes, mode: Optional[Any] = None) -> None:
-                    # type: (bytes, Optional[Any]) -> None
                     """Check the relative path.
                     path may contain a pattern (e.g. foodir/**.txt)"""
                                 raise error.Abort(msg % (path, pycompat.bytestr(prefix)))
                     return True
-                def check(self, path):
+                def check(self, path: bytes) -> bool:
-                    # type: (bytes) -> bool
                     try:
                         self(path)
                         return True
                             self._cached = False
-            def canonpath(root, cwd, myname, auditor=None):
+            def canonpath(
-                # type: (bytes, bytes, bytes, Optional[pathauditor]) -> bytes
+                root: bytes,
+                cwd: bytes,
+                myname: bytes,
+                auditor: Optional[pathauditor] = None,
+            ) -> bytes:
                 """return the canonical path of myname, given cwd and root
                 >>> def check(root, cwd, myname):
                     )
-            def normasprefix(path):
+            def normasprefix(path: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 """normalize the specified path as path prefix
                 Returned value can be used safely for "p.startswith(prefix)",
                     return path
-            def finddirs(path):
+            def finddirs(path: bytes) -> Iterator[bytes]:
-                # type: (bytes) -> Iterator[bytes]
                 pos = path.rfind(b'/')
                 while pos != -1:
                     yield path[:pos]
                         for f in map:
                             addpath(f)
-                def addpath(self, path):
+                def addpath(self, path: bytes) -> None:
-                    # type: (bytes) -> None
                     dirs = self._dirs
                     for base in finddirs(path):
                         if base.endswith(b'/'):
                             return
                         dirs[base] = 1
-                def delpath(self, path):
+                def delpath(self, path: bytes) -> None:
-                    # type: (bytes) -> None
                     dirs = self._dirs
                     for base in finddirs(path):
                         if dirs[base] > 1:
                 def __iter__(self):
                     return iter(self._dirs)
-                def __contains__(self, d):
+                def __contains__(self, d: bytes) -> bool:
-                    # type: (bytes) -> bool
                     return d in self._dirs

mercurial/phases.py

0 +41 -34

             no_bundle_phases = all_internal_phases
-            def supportinternal(repo):
+            def supportinternal(repo: "localrepo.localrepository") -> bool:
-                # type: (localrepo.localrepository) -> bool
                 """True if the internal phase can be used on a repository"""
                 return requirements.INTERNAL_PHASE_REQUIREMENT in repo.requirements
-            def supportarchived(repo):
+            def supportarchived(repo: "localrepo.localrepository") -> bool:
-                # type: (localrepo.localrepository) -> bool
                 """True if the archived phase can be used on a repository"""
                 return requirements.ARCHIVED_PHASE_REQUIREMENT in repo.requirements
-            def _readroots(repo, phasedefaults=None):
+            def _readroots(
-                # type: (localrepo.localrepository, Optional[Phasedefaults]) -> Tuple[Phaseroots, bool]
+                repo: "localrepo.localrepository",
+                phasedefaults: Optional["Phasedefaults"] = None,
+            ) -> Tuple[Phaseroots, bool]:
                 """Read phase roots from disk
                 phasedefaults is a list of fn(repo, roots) callable, which are
                 return roots, dirty
-            def binaryencode(phasemapping):
+            def binaryencode(phasemapping: Dict[int, List[bytes]]) -> bytes:
-                # type: (Dict[int, List[bytes]]) -> bytes
                 """encode a 'phase -> nodes' mapping into a binary stream
                 The revision lists are encoded as (phase, root) pairs.
                 return b''.join(binarydata)
-            def binarydecode(stream):
+            def binarydecode(stream) -> Dict[int, List[bytes]]:
-                # type: (...) -> Dict[int, List[bytes]]
                 """decode a binary stream into a 'phase -> nodes' mapping
                 The (phase, root) pairs are turned back into a dictionary with
             class phasecache:
-                def __init__(self, repo, phasedefaults, _load=True):
+                def __init__(
-                    # type: (localrepo.localrepository, Optional[Phasedefaults], bool) -> None
+                    self,
+                    repo: "localrepo.localrepository",
+                    phasedefaults: Optional["Phasedefaults"],
+                    _load: bool = True,
+                ):
                     if _load:
                         # Cheap trick to allow shallow-copy without copy module
                         self.phaseroots, self.dirty = _readroots(repo, phasedefaults)
                         self.filterunknown(repo)
                         self.opener = repo.svfs
-                def hasnonpublicphases(self, repo):
+                def hasnonpublicphases(self, repo: "localrepo.localrepository") -> bool:
-                    # type: (localrepo.localrepository) -> bool
                     """detect if there are revisions with non-public phase"""
                     repo = repo.unfiltered()
                     cl = repo.changelog
                         revs for phase, revs in self.phaseroots.items() if phase != public
                     )
-                def nonpublicphaseroots(self, repo):
+                def nonpublicphaseroots(
-                    # type: (localrepo.localrepository) -> Set[bytes]
+                    self, repo: "localrepo.localrepository"
+                ) -> Set[bytes]:
                     """returns the roots of all non-public phases
                     The roots are not minimized, so if the secret revisions are
                         ]
                     )
-                def getrevset(self, repo, phases, subset=None):
+                def getrevset(
-                    # type: (localrepo.localrepository, Iterable[int], Optional[Any]) -> Any
+                    self,
+                    repo: "localrepo.localrepository",
+                    phases: Iterable[int],
+                    subset: Optional[Any] = None,
+                ) -> Any:
                     # TODO: finish typing this
                     """return a smartset for the given phases"""
                     self.loadphaserevs(repo)  # ensure phase's sets are loaded
                             self._phasesets[phase] = ps
                     self._loadedrevslen = len(cl)
-                def loadphaserevs(self, repo):
+                def loadphaserevs(self, repo: "localrepo.localrepository") -> None:
-                    # type: (localrepo.localrepository) -> None
                     """ensure phase information is loaded in the object"""
                     if self._phasesets is None:
                         try:
                     self._loadedrevslen = 0
                     self._phasesets = None
-                def phase(self, repo, rev):
+                def phase(self, repo: "localrepo.localrepository", rev: int) -> int:
-                    # type: (localrepo.localrepository, int) -> int
                     # We need a repo argument here to be able to build _phasesets
                     # if necessary. The repository instance is not stored in
                     # phasecache to avoid reference cycles. The changelog instance
                         return True
                     return False
-                def filterunknown(self, repo):
+                def filterunknown(self, repo: "localrepo.localrepository") -> None:
-                    # type: (localrepo.localrepository) -> None
                     """remove unknown nodes from the phase boundary
                     Nothing is lost as unknown nodes only hold data for their descendants.
                 repo._phasecache.replace(phcache)
-            def listphases(repo):
+            def listphases(repo: "localrepo.localrepository") -> Dict[bytes, bytes]:
-                # type: (localrepo.localrepository) -> Dict[bytes, bytes]
                 """List phases root for serialization over pushkey"""
                 # Use ordered dictionary so behavior is deterministic.
                 keys = util.sortdict()
                 return keys
-            def pushphase(repo, nhex, oldphasestr, newphasestr):
+            def pushphase(
-                # type: (localrepo.localrepository, bytes, bytes, bytes) -> bool
+                repo: "localrepo.localrepository",
+                nhex: bytes,
+                oldphasestr: bytes,
+                newphasestr: bytes,
+            ) -> bool:
                 """List phases root for serialization over pushkey"""
                 repo = repo.unfiltered()
                 with repo.lock():
                 return pycompat.maplist(cl.node, sorted(new_heads))
-            def newcommitphase(ui):
+            def newcommitphase(ui: "uimod.ui") -> int:
-                # type: (uimod.ui) -> int
                 """helper to get the target phase of new commit
                 Handle all possible values for the phases.new-commit options.
                     )
-            def hassecret(repo):
+            def hassecret(repo: "localrepo.localrepository") -> bool:
-                # type: (localrepo.localrepository) -> bool
                 """utility function that check if a repo have any secret changeset."""
                 return bool(repo._phasecache.phaseroots[secret])
-            def preparehookargs(node, old, new):
+            def preparehookargs(
-                # type: (bytes, Optional[int], Optional[int]) -> Dict[bytes, bytes]
+                node: bytes,
+                old: Optional[int],
+                new: Optional[int],
+            ) -> Dict[bytes, bytes]:
                 if old is None:
                     old = b''
                 else:

mercurial/pvec.py

0 +1 -2

                 return v
-            def _str(v, l):
+            def _str(v: int, l: int) -> bytes:
-                # type: (int, int) -> bytes
                 bs = b""
                 for p in range(l):
                     bs = pycompat.bytechr(v & 255) + bs

mercurial/state.py

0 +1 -2

                     self._repo = repo
                     self.fname = fname
-                def read(self):
+                def read(self) -> Dict[bytes, Any]:
-                    # type: () -> Dict[bytes, Any]
                     """read the existing state file and return a dict of data stored"""
                     return self._read()

mercurial/subrepoutil.py

0 +26 -18

             Substate = Dict[bytes, Tuple[bytes, bytes, bytes]]
-            def state(ctx, ui):
+            def state(ctx: "context.changectx", ui: "uimod.ui") -> Substate:
-                # type: (context.changectx, uimod.ui) -> Substate
                 """return a state dict, mapping subrepo paths configured in .hgsub
                 to tuple: (source from .hgsub, revision from .hgsubstate, kind
                 (key in types dict))
                     except FileNotFoundError:
                         pass
-                def remap(src):
+                def remap(src: bytes) -> bytes:
-                    # type: (bytes) -> bytes
                     for pattern, repl in p.items(b'subpaths'):
                         # Turn r'C:\foo\bar' into r'C:\\foo\\bar' since re.sub
                         # does a string decode.
                 return state
-            def writestate(repo, state):
+            def writestate(repo: "localrepo.localrepository", state: Substate) -> None:
-                # type: (localrepo.localrepository, Substate) -> None
                 """rewrite .hgsubstate in (outer) repo with these subrepo states"""
                 lines = [
                     b'%s %s\n' % (state[s][1], s)
                 repo.wwrite(b'.hgsubstate', b''.join(lines), b'')
-            def submerge(repo, wctx, mctx, actx, overwrite, labels=None):
+            def submerge(
-                # type: (localrepo.localrepository, context.workingctx, context.changectx, context.changectx, bool, Optional[Any]) -> Substate
+                repo: "localrepo.localrepository",
+                wctx: "context.workingctx",
+                mctx: "context.changectx",
+                actx: "context.changectx",
+                overwrite: bool,
+                labels: Optional[Any] = None,
+            ) -> Substate:
                 # TODO: type the `labels` arg
                 """delegated from merge.applyupdates: merging of .hgsubstate file
                 in working context, merging context and ancestor context"""
                 return sm
-            def precommit(ui, wctx, status, match, force=False):
+            def precommit(
-                # type: (uimod.ui, context.workingcommitctx, scmutil.status, matchmod.basematcher, bool) -> Tuple[List[bytes], Set[bytes], Substate]
+                ui: "uimod.ui",
+                wctx: "context.workingcommitctx",
+                status: "scmutil.status",
+                match: "matchmod.basematcher",
+                force: bool = False,
+            ) -> Tuple[List[bytes], Set[bytes], Substate]:
                 """Calculate .hgsubstate changes that should be applied before committing
                 Returns (subs, commitsubs, newstate) where
                 return posixpath.normpath(path)
-            def reporelpath(repo):
+            def reporelpath(repo: "localrepo.localrepository") -> bytes:
-                # type: (localrepo.localrepository) -> bytes
                 """return path to this (sub)repo as seen from outermost repo"""
                 parent = repo
                 while hasattr(parent, '_subparent'):
                 return repo.root[len(pathutil.normasprefix(parent.root)) :]
-            def subrelpath(sub):
+            def subrelpath(sub: "subrepo.abstractsubrepo") -> bytes:
-                # type: (subrepo.abstractsubrepo) -> bytes
                 """return path to this subrepo as seen from outermost repo"""
                 return sub._relpath
-            def _abssource(repo, push=False, abort=True):
+            def _abssource(
-                # type: (localrepo.localrepository, bool, bool) -> Optional[bytes]
+                repo: "localrepo.localrepository",
+                push: bool = False,
+                abort: bool = True,
+            ) -> Optional[bytes]:
                 """return pull/push path of repo - either based on parent repo .hgsub info
                 or on the top repo config. Abort or return None if no source found."""
                 if hasattr(repo, '_subparent'):
                     raise error.Abort(_(b"default path for subrepository not found"))
-            def newcommitphase(ui, ctx):
+            def newcommitphase(ui: "uimod.ui", ctx: "context.changectx") -> int:
-                # type: (uimod.ui, context.changectx) -> int
                 commitphase = phases.newcommitphase(ui)
                 substate = getattr(ctx, "substate", None)
                 if not substate:

mercurial/util.py

0 +26 -44

             username = platform.username
-            def setumask(val):
+            def setumask(val: int) -> None:
-                # type: (int) -> None
                 '''updates the umask. used by chg server'''
                 if pycompat.iswindows:
                     return
                 nogc = lambda x: x
-            def pathto(root, n1, n2):
+            def pathto(root: bytes, n1: bytes, n2: bytes) -> bytes:
-                # type: (bytes, bytes, bytes) -> bytes
                 """return the relative path from one place to another.
                 root should use os.sep to separate directories
                 n1 should use os.sep to separate directories
             _winreservedchars = b':*?"<>|'
-            def checkwinfilename(path):
+            def checkwinfilename(path: bytes) -> Optional[bytes]:
-                # type: (bytes) -> Optional[bytes]
                 r"""Check that the base-relative path is a valid filename on Windows.
                 Returns None if the path is ok, or a UI string describing the problem.
                 os.close(ld)
-            def readlock(pathname):
+            def readlock(pathname: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 try:
                     return readlink(pathname)
                 except OSError as why:
             # File system features
-            def fscasesensitive(path):
+            def fscasesensitive(path: bytes) -> bool:
-                # type: (bytes) -> bool
                 """
                 Return true if the given path is on a case-sensitive filesystem
             _fspathcache = {}
-            def fspath(name, root):
+            def fspath(name: bytes, root: bytes) -> bytes:
-                # type: (bytes, bytes) -> bytes
                 """Get name in the case stored in the filesystem
                 The name should be relative to root, and be normcase-ed for efficiency.
                 return b''.join(result)
-            def checknlink(testfile):
+            def checknlink(testfile: bytes) -> bool:
-                # type: (bytes) -> bool
                 '''check whether hardlink count reporting works properly'''
                 # testfile may be open, so we need a separate file for checking to
                             pass
-            def endswithsep(path):
+            def endswithsep(path: bytes) -> bool:
-                # type: (bytes) -> bool
                 '''Check path ends with os.sep or os.altsep.'''
                 return bool(  # help pytype
                     path.endswith(pycompat.ossep)
                 )
-            def splitpath(path):
+            def splitpath(path: bytes) -> List[bytes]:
-                # type: (bytes) -> List[bytes]
                 """Split path by os.sep.
                 Note that this function does not use os.altsep because this is
                 an alternative of simple "xxx.split(os.sep)".
                         raise
-            def unlinkpath(f, ignoremissing=False, rmdir=True):
+            def unlinkpath(
-                # type: (bytes, bool, bool) -> None
+                f: bytes, ignoremissing: bool = False, rmdir: bool = True
+            ) -> None:
                 """unlink and remove the directory if it is empty"""
                 if ignoremissing:
                     tryunlink(f)
                         pass
-            def tryunlink(f):
+            def tryunlink(f: bytes) -> None:
-                # type: (bytes) -> None
                 """Attempt to remove a file, ignoring FileNotFoundError."""
                 try:
                     unlink(f)
                     pass
-            def makedirs(name, mode=None, notindexed=False):
+            def makedirs(
-                # type: (bytes, Optional[int], bool) -> None
+                name: bytes, mode: Optional[int] = None, notindexed: bool = False
+            ) -> None:
                 """recursive directory creation with parent mode inheritance
                 Newly created directories are marked as "not to be indexed by
                     os.chmod(name, mode)
-            def readfile(path):
+            def readfile(path: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 with open(path, b'rb') as fp:
                     return fp.read()
-            def writefile(path, text):
+            def writefile(path: bytes, text: bytes) -> None:
-                # type: (bytes, bytes) -> None
                 with open(path, b'wb') as fp:
                     fp.write(text)
-            def appendfile(path, text):
+            def appendfile(path: bytes, text: bytes) -> None:
-                # type: (bytes, bytes) -> None
                 with open(path, b'ab') as fp:
                     fp.write(text)
                 return go
-            def processlinerange(fromline, toline):
+            def processlinerange(fromline: int, toline: int) -> Tuple[int, int]:
-                # type: (int, int) -> Tuple[int, int]
                 """Check that linerange <fromline>:<toline> makes sense and return a
 -based range.
             _eolre = remod.compile(br'\r*\n')
-            def tolf(s):
+            def tolf(s: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 return _eolre.sub(b'\n', s)
-            def tocrlf(s):
+            def tocrlf(s: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 return _eolre.sub(b'\r\n', s)
                 return fp
-            def iterlines(iterator):
+            def iterlines(iterator: Iterable[bytes]) -> Iterator[bytes]:
-                # type: (Iterable[bytes]) -> Iterator[bytes]
                 for chunk in iterator:
                     for line in chunk.splitlines():
                         yield line
-            def expandpath(path):
+            def expandpath(path: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 return os.path.expanduser(os.path.expandvars(path))
             )
-            def sizetoint(s):
+            def sizetoint(s: bytes) -> int:
-                # type: (bytes) -> int
                 """Convert a space specifier to a byte count.
                 >>> sizetoint(b'30')
                     yield
-            def _estimatememory():
+            def _estimatememory() -> Optional[int]:
-                # type: () -> Optional[int]
                 """Provide an estimate for the available system memory in Bytes.
                 If no estimate can be provided on the platform, returns None.

mercurial/utils/dateutil.py

0 +20 -18

             )
-            def makedate(timestamp=None):
+            def makedate(timestamp: Optional[float] = None) -> hgdate:
-                # type: (Optional[float]) -> hgdate
                 """Return a unix timestamp (or the current time) as a (unixtime,
                 offset) tuple based off the local timezone."""
                 if timestamp is None:
                 return timestamp, tz
-            def datestr(date=None, format=b'%a %b %d %H:%M:%S %Y %1%2'):
+            def datestr(
-                # type: (Optional[hgdate], bytes) -> bytes
+                date: Optional[hgdate] = None,
+                format: bytes = b'%a %b %d %H:%M:%S %Y %1%2',
+            ) -> bytes:
                 """represent a (unixtime, offset) tuple as a localized time.
                 unixtime is seconds since the epoch, and offset is the time zone's
                 number of seconds away from UTC.
                 return s
-            def shortdate(date=None):
+            def shortdate(date: Optional[hgdate] = None) -> bytes:
-                # type: (Optional[hgdate]) -> bytes
                 """turn (timestamp, tzoff) tuple into iso 8631 date."""
                 return datestr(date, format=b'%Y-%m-%d')
-            def parsetimezone(s):
+            def parsetimezone(s: bytes) -> Tuple[Optional[int], bytes]:
-                # type: (bytes) -> Tuple[Optional[int], bytes]
                 """find a trailing timezone, if any, in string, and return a
                 (offset, remainder) pair"""
                 s = pycompat.bytestr(s)
                 return None, s
-            def strdate(string, format, defaults=None):
+            def strdate(
-                # type: (bytes, bytes, Optional[Dict[bytes, Tuple[bytes, bytes]]]) -> hgdate
+                string: bytes,
+                format: bytes,
+                defaults: Optional[Dict[bytes, Tuple[bytes, bytes]]] = None,
+            ) -> hgdate:
                 """parse a localized time string and return a (unixtime, offset) tuple.
                 if the string cannot be parsed, ValueError is raised."""
                 if defaults is None:
                 return unixtime, offset
-            def parsedate(date, formats=None, bias=None):
+            def parsedate(
-                # type: (Union[bytes, hgdate], Optional[Iterable[bytes]], Optional[Dict[bytes, bytes]]) -> hgdate
+                date: Union[bytes, hgdate],
+                formats: Optional[Iterable[bytes]] = None,
+                bias: Optional[Dict[bytes, bytes]] = None,
+            ) -> hgdate:
                 """parse a localized date/time and return a (unixtime, offset) tuple.
                 The date may be a "unixtime offset" string or in one of the specified
                 return when, offset
-            def matchdate(date):
+            def matchdate(date: bytes) -> Callable[[float], bool]:
-                # type: (bytes) -> Callable[[float], bool]
                 """Return a function that matches a given date match specifier
                 Formats include:
                 False
                 """
-                def lower(date):
+                def lower(date: bytes) -> float:
-                    # type: (bytes) -> float
                     d = {b'mb': b"1", b'd': b"1"}
                     return parsedate(date, extendeddateformats, d)[0]
-                def upper(date):
+                def upper(date: bytes) -> float:
-                    # type: (bytes) -> float
                     d = {b'mb': b"12", b'HI': b"23", b'M': b"59", b'S': b"59"}
                     for days in (b"31", b"30", b"29"):
                         try:

mercurial/utils/urlutil.py

0 +14 -18

             urlreq = urllibcompat.urlreq
-            def getport(port):
+            def getport(port: Union[bytes, int]) -> int:
-                # type: (Union[bytes, int]) -> int
                 """Return the port for a given network service.
                 If port is an integer, it's returned as is. If it's a string, it's
                 _safepchars = b"/!~*'()+:\\"
                 _matchscheme = remod.compile(b'^[a-zA-Z0-9+.\\-]+:').match
-                def __init__(self, path, parsequery=True, parsefragment=True):
+                def __init__(
-                    # type: (bytes, bool, bool) -> None
+                    self,
+                    path: bytes,
+                    parsequery: bool = True,
+                    parsefragment: bool = True,
+                ) -> None:
                     # We slowly chomp away at path until we have only the path left
                     self.scheme = self.user = self.passwd = self.host = None
                     self.port = self.path = self.query = self.fragment = None
                         return True  # POSIX-style
                     return False
-                def localpath(self):
+                def localpath(self) -> bytes:
-                    # type: () -> bytes
                     if self.scheme == b'file' or self.scheme == b'bundle':
                         path = self.path or b'/'
                         # For Windows, we need to promote hosts containing drive
                     )
-            def hasscheme(path):
+            def hasscheme(path: bytes) -> bool:
-                # type: (bytes) -> bool
                 return bool(url(path).scheme)  # cast to help pytype
-            def hasdriveletter(path):
+            def hasdriveletter(path: bytes) -> bool:
-                # type: (bytes) -> bool
                 return bool(path) and path[1:2] == b':' and path[0:1].isalpha()
-            def urllocalpath(path):
+            def urllocalpath(path: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 return url(path, parsequery=False, parsefragment=False).localpath()
-            def checksafessh(path):
+            def checksafessh(path: bytes) -> None:
-                # type: (bytes) -> None
                 """check if a path / url is a potentially unsafe ssh exploit (SEC)
                 This is a sanity check for ssh urls. ssh will parse the first item as
                     )
-            def hidepassword(u):
+            def hidepassword(u: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 '''hide user credential in a url string'''
                 u = url(u)
                 if u.passwd:
                 return bytes(u)
-            def removeauth(u):
+            def removeauth(u: bytes) -> bytes:
-                # type: (bytes) -> bytes
                 '''remove all authentication information from a url string'''
                 u = url(u)
                 u.user = u.passwd = None

General Comments 0

Write
Preview

You need to be logged in to leave comments. Login now

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages