mercurial-scm/hg: mercurial/shelve.py comparison

comparison mercurial/shelve.py @ 51812:a1a94d488e14

typing: add type hints to `mercurial.shelve` Pytype wasn't flagging anything here yet, but PyCharm was really unhappy about the usage of `state` objects being passed to various methods that accessed attrs on it, without any obvious attrs on the class because there's no contructor. Filling that out made PyCharm happy, and a few other things needed to be filled in to make that easier, so I made a pass over the whole file and filled in the trivial hints. The other repo, ui, context, matcher, and pats items can be filled in after the context and match modules are typed.

author	Matt Harbison <matt_harbison@yahoo.com>
date	Tue, 20 Aug 2024 22:34:51 -0400
parents	54b1a3738530
children	adbfbbf9963f

comparison

equal deleted inserted replaced

-:460e80488cf0
+:a1a94d488e14
 import collections
 import io
 import itertools
 import stat
+from typing import (
+Any,
+Dict,
+Iterable,
+Iterator,
+List,
+Sequence,
+Tuple,
+)
 from .i18n import _
 from .node import (
 bin,
 hex,
 from . import (
 bookmarks,
 bundle2,
 changegroup,
 cmdutil,
+context as contextmod,
 discovery,
 error,
 exchange,
 hg,
 lock as lockmod,
 # generic user for all shelve operations
 shelveuser = b'shelve@localhost'
 class ShelfDir:
-def __init__(self, repo, for_backups=False):
+def __init__(self, repo, for_backups: bool = False) -> None:
 if for_backups:
 self.vfs = vfsmod.vfs(repo.vfs.join(backupdir))
 else:
 self.vfs = vfsmod.vfs(repo.vfs.join(shelvedir))
-def get(self, name):
+def get(self, name: bytes) -> "Shelf":
 return Shelf(self.vfs, name)
-def listshelves(self):
+def listshelves(self) -> List[Tuple[float, bytes]]:
 """return all shelves in repo as list of (time, name)"""
 try:
 names = self.vfs.listdir()
 except FileNotFoundError:
 return []
 mtime = shelf.mtime()
 info.append((mtime, name))
 return sorted(info, reverse=True)
-def _use_internal_phase(repo):
+def _use_internal_phase(repo) -> bool:
 return (
 phases.supportinternal(repo)
 and repo.ui.config(b'shelve', b'store') == b'internal'
 )
-def _target_phase(repo):
+def _target_phase(repo) -> int:
 return phases.internal if _use_internal_phase(repo) else phases.secret
 class Shelf:
 """Represents a shelf, including possibly multiple files storing it.
 Old shelves will have a .patch and a .hg file. Newer shelves will
 also have a .shelve file. This class abstracts away some of the
 differences and lets you work with the shelf as a whole.
 """
-def __init__(self, vfs, name):
+def __init__(self, vfs: vfsmod.vfs, name: bytes) -> None:
 self.vfs = vfs
 self.name = name
-def exists(self):
+def exists(self) -> bool:
 return self._exists(b'.shelve') or self._exists(b'.patch', b'.hg')
-def _exists(self, *exts):
+def _exists(self, *exts: bytes) -> bool:
 return all(self.vfs.exists(self.name + ext) for ext in exts)
-def mtime(self):
+def mtime(self) -> float:
 try:
 return self._stat(b'.shelve')[stat.ST_MTIME]
 except FileNotFoundError:
 return self._stat(b'.patch')[stat.ST_MTIME]
-def _stat(self, ext):
+def _stat(self, ext: bytes):
 return self.vfs.stat(self.name + ext)
-def writeinfo(self, info):
+def writeinfo(self, info) -> None:
 scmutil.simplekeyvaluefile(self.vfs, self.name + b'.shelve').write(info)
-def hasinfo(self):
+def hasinfo(self) -> bool:
 return self.vfs.exists(self.name + b'.shelve')
 def readinfo(self):
 return scmutil.simplekeyvaluefile(
 self.vfs, self.name + b'.shelve'
 ).read()
-def writebundle(self, repo, bases, node):
+def writebundle(self, repo, bases, node) -> None:
 cgversion = changegroup.safeversion(repo)
 if cgversion == b'01':
 btype = b'HG10BZ'
 compression = None
 else:
 btype,
 self.vfs,
 compression=compression,
 )
-def applybundle(self, repo, tr):
+def applybundle(self, repo, tr) -> contextmod.changectx:
 filename = self.name + b'.hg'
 fp = self.vfs(filename)
 try:
 targetphase = _target_phase(repo)
 gen = exchange.readbundle(repo.ui, fp, filename, self.vfs)
 shelvectx = repo[shelverev]
 return shelvectx
 finally:
 fp.close()
-def open_patch(self, mode=b'rb'):
+def open_patch(self, mode: bytes = b'rb'):
 return self.vfs(self.name + b'.patch', mode)
-def patch_from_node(self, repo, node):
+def patch_from_node(self, repo, node) -> io.BytesIO:
 repo = repo.unfiltered()
 match = _optimized_match(repo, node)
 fp = io.BytesIO()
 cmdutil.exportfile(
 repo,
 # prefer node-based shelf
 return self.patch_from_node(repo, self.readinfo()[b'node'])
 except (FileNotFoundError, error.RepoLookupError):
 return self.open_patch()
-def _backupfilename(self, backupvfs, filename):
+def _backupfilename(self, backupvfs: vfsmod.vfs, filename: bytes) -> bytes:
-def gennames(base):
+def gennames(base: bytes):
 yield base
 base, ext = base.rsplit(b'.', 1)
 for i in itertools.count(1):
 yield b'%s-%d.%s' % (base, i, ext)
 for n in gennames(filename):
 if not backupvfs.exists(n):
 return backupvfs.join(n)
-def movetobackup(self, backupvfs):
+# Help pytype- gennames() yields infinitely
+raise error.ProgrammingError("unreachable")
+def movetobackup(self, backupvfs: vfsmod.vfs) -> None:
 if not backupvfs.isdir():
 backupvfs.makedir()
 for suffix in shelvefileextensions:
 filename = self.name + b'.' + suffix
 if self.vfs.exists(filename):
 util.rename(
 self.vfs.join(filename),
 self._backupfilename(backupvfs, filename),
 )
-def delete(self):
+def delete(self) -> None:
 for ext in shelvefileextensions:
 self.vfs.tryunlink(self.name + b'.' + ext)
 def changed_files(self, ui, repo):
 try:
 except (FileNotFoundError, error.RepoLookupError):
 filename = self.vfs.join(self.name + b'.patch')
 return patch.changedfiles(ui, repo, filename)
-def _optimized_match(repo, node):
+def _optimized_match(repo, node: bytes):
 """
 Create a matcher so that prefetch doesn't attempt to fetch
 the entire repository pointlessly, and as an optimisation
 for movedirstate, if needed.
 """
 Handles saving and restoring a shelved state. Ensures that different
 versions of a shelved state are possible and handles them appropriately.
 """
+# Class-wide constants
 _version = 2
 _filename = b'shelvedstate'
 _keep = b'keep'
 _nokeep = b'nokeep'
 # colon is essential to differentiate from a real bookmark name
 _noactivebook = b':no-active-bookmark'
 _interactive = b'interactive'
+# Per instance attrs
+name: bytes
+wctx: contextmod.workingctx
+pendingctx: contextmod.changectx
+parents: List[bytes]
+nodestoremove: List[bytes]
+branchtorestore: bytes
+keep: bool
+activebookmark: bytes
+interactive: bool
 @classmethod
-def _verifyandtransform(cls, d):
+def _verifyandtransform(cls, d: Dict[bytes, Any]) -> None:
 """Some basic shelvestate syntactic verification and transformation"""
 try:
 d[b'originalwctx'] = bin(d[b'originalwctx'])
 d[b'pendingctx'] = bin(d[b'pendingctx'])
 d[b'parents'] = [bin(h) for h in d[b'parents'].split(b' ') if h]
 ]
 except (ValueError, KeyError) as err:
 raise error.CorruptedState(stringutil.forcebytestr(err))
 @classmethod
-def _getversion(cls, repo):
+def _getversion(cls, repo) -> int:
 """Read version information from shelvestate file"""
 fp = repo.vfs(cls._filename)
 try:
 version = int(fp.readline().strip())
 except ValueError as err:
 finally:
 fp.close()
 return version
 @classmethod
-def _readold(cls, repo):
+def _readold(cls, repo) -> Dict[bytes, Any]:
 """Read the old position-based version of a shelvestate file"""
 # Order is important, because old shelvestate file uses it
 # to detemine values of fields (i.g. name is on the second line,
 # originalwctx is on the third and so forth). Please do not change.
 keys = [
 @classmethod
 def save(
 cls,
 repo,
-name,
+name: bytes,
-originalwctx,
+originalwctx: contextmod.workingctx,
-pendingctx,
+pendingctx: contextmod.changectx,
-nodestoremove,
+nodestoremove: List[bytes],
-branchtorestore,
+branchtorestore: bytes,
-keep=False,
+keep: bool = False,
-activebook=b'',
+activebook: bytes = b'',
-interactive=False,
+interactive: bool = False,
-):
+) -> None:
 info = {
 b"name": name,
 b"originalwctx": hex(originalwctx.node()),
 b"pendingctx": hex(pendingctx.node()),
 b"parents": b' '.join([hex(p) for p in repo.dirstate.parents()]),
 scmutil.simplekeyvaluefile(repo.vfs, cls._filename).write(
 info, firstline=(b"%d" % cls._version)
 )
 @classmethod
-def clear(cls, repo):
+def clear(cls, repo) -> None:
 repo.vfs.unlinkpath(cls._filename, ignoremissing=True)
-def cleanupoldbackups(repo):
+def cleanupoldbackups(repo) -> None:
 maxbackups = repo.ui.configint(b'shelve', b'maxbackups')
 backup_dir = ShelfDir(repo, for_backups=True)
 hgfiles = backup_dir.listshelves()
 if maxbackups > 0 and maxbackups < len(hgfiles):
 bordermtime = hgfiles[maxbackups - 1][0]
 # keep it, because timestamp can't decide exact order of backups
 continue
 backup_dir.get(name).delete()
-def _backupactivebookmark(repo):
+def _backupactivebookmark(repo) -> bytes:
 activebookmark = repo._activebookmark
 if activebookmark:
 bookmarks.deactivate(repo)
 return activebookmark
-def _restoreactivebookmark(repo, mark):
+def _restoreactivebookmark(repo, mark) -> None:
 if mark:
 bookmarks.activate(repo, mark)
-def _aborttransaction(repo, tr):
+def _aborttransaction(repo, tr) -> None:
 """Abort current transaction for shelve/unshelve, but keep dirstate"""
 # disable the transaction invalidation of the dirstate, to preserve the
 # current change in memory.
 ds = repo.dirstate
 # The assert below check that nobody else did such wrapping.
 assert repo.currenttransaction() is None
 repo.dirstate.write(None)
 ds.setbranch(current_branch, None)
-def getshelvename(repo, parent, opts):
+def getshelvename(repo, parent, opts) -> bytes:
 """Decide on the name this shelve is going to have"""
 def gennames():
 yield label
 for i in itertools.count(1):
 break
 return name
-def mutableancestors(ctx):
+def mutableancestors(ctx) -> Iterator[bytes]:
 """return all mutable ancestors for ctx (included)
 Much faster than the revset ancestors(ctx) & draft()"""
 seen = {nullrev}
 visit = collections.deque()
 seen.add(rev)
 if parent.mutable():
 visit.append(parent)
-def getcommitfunc(extra, interactive, editor=False):
+def getcommitfunc(extra, interactive: bool, editor: bool = False):
 def commitfunc(ui, repo, message, match, opts):
 hasmq = hasattr(repo, 'mq')
 if hasmq:
 saved, repo.mq.checkapplied = repo.mq.checkapplied, False
 return commitfunc(ui, repo, message, match, opts)
 return interactivecommitfunc if interactive else commitfunc
-def _nothingtoshelvemessaging(ui, repo, pats, opts):
+def _nothingtoshelvemessaging(ui, repo, pats, opts) -> None:
 stat = repo.status(match=scmutil.match(repo[None], pats, opts))
 if stat.deleted:
 ui.status(
 _(b"nothing changed (%d missing files, see 'hg status')\n")
 % len(stat.deleted)
 )
 else:
 ui.status(_(b"nothing changed\n"))
-def _shelvecreatedcommit(repo, node, name, match):
+def _shelvecreatedcommit(repo, node: bytes, name: bytes, match) -> None:
 info = {b'node': hex(node)}
 shelf = ShelfDir(repo).get(name)
 shelf.writeinfo(info)
 bases = list(mutableancestors(repo[node]))
 shelf.writebundle(repo, bases, node)
 cmdutil.exportfile(
 repo, [node], fp, opts=mdiff.diffopts(git=True), match=match
 )
-def _includeunknownfiles(repo, pats, opts, extra):
+def _includeunknownfiles(repo, pats, opts, extra) -> None:
 s = repo.status(match=scmutil.match(repo[None], pats, opts), unknown=True)
 if s.unknown:
 extra[b'shelve_unknown'] = b'\0'.join(s.unknown)
 repo[None].add(s.unknown)
-def _finishshelve(repo, tr):
+def _finishshelve(repo, tr) -> None:
 if _use_internal_phase(repo):
 tr.close()
 else:
 _aborttransaction(repo, tr)
 finally:
 _restoreactivebookmark(repo, activebookmark)
 lockmod.release(tr, lock)
-def _isbareshelve(pats, opts):
+def _isbareshelve(pats, opts) -> bool:
 return (
 not pats
 and not opts.get(b'interactive', False)
 and not opts.get(b'include', False)
 and not opts.get(b'exclude', False)
 )
-def _iswctxonnewbranch(repo):
+def _iswctxonnewbranch(repo) -> bool:
 return repo[None].branch() != repo[b'.'].branch()
-def cleanupcmd(ui, repo):
+def cleanupcmd(ui, repo) -> None:
 """subcommand that deletes all shelves"""
 with repo.wlock():
 shelf_dir = ShelfDir(repo)
 backupvfs = vfsmod.vfs(repo.vfs.join(backupdir))
 for _mtime, name in shelf_dir.listshelves():
 shelf_dir.get(name).movetobackup(backupvfs)
 cleanupoldbackups(repo)
-def deletecmd(ui, repo, pats):
+def deletecmd(ui, repo, pats) -> None:
 """subcommand that deletes a specific shelve"""
 if not pats:
 raise error.InputError(_(b'no shelved changes specified!'))
 with repo.wlock():
 backupvfs = vfsmod.vfs(repo.vfs.join(backupdir))
 )
 shelf.movetobackup(backupvfs)
 cleanupoldbackups(repo)
-def listcmd(ui, repo, pats, opts):
+def listcmd(ui, repo, pats: Iterable[bytes], opts) -> None:
 """subcommand that displays the list of shelves"""
 pats = set(pats)
 width = 80
 if not ui.plain():
 width = ui.termwidth()
 if opts[b'stat']:
 for chunk, label in patch.diffstatui(difflines, width=width):
 ui.write(chunk, label=label)
-def patchcmds(ui, repo, pats, opts):
+def patchcmds(ui, repo, pats: Sequence[bytes], opts) -> None:
 """subcommand that displays shelves"""
 shelf_dir = ShelfDir(repo)
 if len(pats) == 0:
 shelves = shelf_dir.listshelves()
 if not shelves:
 raise error.Abort(_(b"cannot find shelf %s") % shelfname)
 listcmd(ui, repo, pats, opts)
-def checkparents(repo, state):
+def checkparents(repo, state: shelvedstate) -> None:
 """check parent while resuming an unshelve"""
 if state.parents != repo.dirstate.parents():
 raise error.Abort(
 _(b'working directory parents do not match unshelve state')
 )
-def _loadshelvedstate(ui, repo, opts):
+def _loadshelvedstate(ui, repo, opts) -> shelvedstate:
 try:
 state = shelvedstate.load(repo)
 if opts.get(b'keep') is None:
 opts[b'keep'] = state.keep
 return state
 raise error.ProgrammingError(
 "a corrupted shelvedstate exists without --abort or --continue"
 )
-def unshelveabort(ui, repo, state):
+def unshelveabort(ui, repo, state: shelvedstate) -> None:
 """subcommand that abort an in-progress unshelve"""
 with repo.lock():
 try:
 checkparents(repo, state)
 finally:
 shelvedstate.clear(repo)
 ui.warn(_(b"unshelve of '%s' aborted\n") % state.name)
-def hgabortunshelve(ui, repo):
+def hgabortunshelve(ui, repo) -> None:
 """logic to  abort unshelve using 'hg abort"""
 with repo.wlock():
 state = _loadshelvedstate(ui, repo, {b'abort': True})
 return unshelveabort(ui, repo, state)
-def mergefiles(ui, repo, wctx, shelvectx):
+def mergefiles(ui, repo, wctx, shelvectx) -> None:
 """updates to wctx and merges the changes from shelvectx into the
 dirstate."""
 with ui.configoverride({(b'ui', b'quiet'): True}):
 hg.update(repo, wctx.node())
 cmdutil.revert(ui, repo, shelvectx)
-def restorebranch(ui, repo, branchtorestore):
+def restorebranch(ui, repo, branchtorestore: bytes) -> None:
 if branchtorestore and branchtorestore != repo.dirstate.branch():
 repo.dirstate.setbranch(branchtorestore, repo.currenttransaction())
 ui.status(
 _(b'marked working directory as branch %s\n') % branchtorestore
 )
-def unshelvecleanup(ui, repo, name, opts):
+def unshelvecleanup(ui, repo, name: bytes, opts) -> None:
 """remove related files after an unshelve"""
 if not opts.get(b'keep'):
 backupvfs = vfsmod.vfs(repo.vfs.join(backupdir))
 ShelfDir(repo).get(name).movetobackup(backupvfs)
 cleanupoldbackups(repo)
-def unshelvecontinue(ui, repo, state, opts):
+def unshelvecontinue(ui, repo, state: shelvedstate, opts) -> None:
 """subcommand to continue an in-progress unshelve"""
 # We're finishing off a merge. First parent is our original
 # parent, second is the temporary "fake" commit we're unshelving.
 interactive = state.interactive
 basename = state.name
 unshelvecleanup(ui, repo, state.name, opts)
 _restoreactivebookmark(repo, state.activebookmark)
 ui.status(_(b"unshelve of '%s' complete\n") % state.name)
-def hgcontinueunshelve(ui, repo):
+def hgcontinueunshelve(ui, repo) -> None:
 """logic to resume unshelve using 'hg continue'"""
 with repo.wlock():
 state = _loadshelvedstate(ui, repo, {b'continue': True})
 return unshelvecontinue(ui, repo, state, {b'keep': state.keep})
 node = cmdutil.commit(ui, repo, commitfunc, [], tempopts)
 tmpwctx = repo[node]
 return tmpwctx, addedbefore
-def _unshelverestorecommit(ui, repo, tr, basename):
+def _unshelverestorecommit(ui, repo, tr, basename: bytes):
 """Recreate commit in the repository during the unshelve"""
 repo = repo.unfiltered()
 node = None
 shelf = ShelfDir(repo).get(basename)
 if shelf.hasinfo():
 shelvectx = repo[node]
 return repo, shelvectx
-def _createunshelvectx(ui, repo, shelvectx, basename, interactive, opts):
+def _createunshelvectx(
+ui, repo, shelvectx, basename: bytes, interactive: bool, opts
+) -> Tuple[bytes, bool]:
 """Handles the creation of unshelve commit and updates the shelve if it
 was partially unshelved.
 If interactive is:
 ui,
 repo,
 opts,
 tr,
 oldtiprev,
-basename,
+basename: bytes,
 pctx,
 tmpwctx,
 shelvectx,
 branchtorestore,
 activebookmark,
 merge.update(tmpwctx)
 return shelvectx, ispartialunshelve
-def _forgetunknownfiles(repo, shelvectx, addedbefore):
+def _forgetunknownfiles(repo, shelvectx, addedbefore) -> None:
 # Forget any files that were unknown before the shelve, unknown before
 # unshelve started, but are now added.
 shelveunknown = shelvectx.extra().get(b'shelve_unknown')
 if not shelveunknown:
 return
 addedafter = frozenset(repo.status().added)
 toforget = (addedafter & shelveunknown) - addedbefore
 repo[None].forget(toforget)
-def _finishunshelve(repo, oldtiprev, tr, activebookmark):
+def _finishunshelve(repo, oldtiprev, tr, activebookmark) -> None:
 _restoreactivebookmark(repo, activebookmark)
 # We used to manually strip the commit to update inmemory structure and
 # prevent some issue around hooks. This no longer seems to be the case, so
 # we simply abort the transaction.
 _aborttransaction(repo, tr)
-def _checkunshelveuntrackedproblems(ui, repo, shelvectx):
+def _checkunshelveuntrackedproblems(ui, repo, shelvectx) -> None:
 """Check potential problems which may result from working
 copy having untracked changes."""
 wcdeleted = set(repo.status().deleted)
 shelvetouched = set(shelvectx.files())
 intersection = wcdeleted.intersection(shelvetouched)
 m = _(b"shelved change touches missing files")
 hint = _(b"run hg status to see which files are missing")
 raise error.Abort(m, hint=hint)
-def unshelvecmd(ui, repo, *shelved, **opts):
+def unshelvecmd(ui, repo, *shelved, **opts) -> None:
 opts = pycompat.byteskwargs(opts)
 abortf = opts.get(b'abort')
 continuef = opts.get(b'continue')
 interactive = opts.get(b'interactive')
 if not abortf and not continuef:
 raise error.InputError(
 _(b'cannot use both continue and interactive')
 )
 elif continuef:
 return unshelvecontinue(ui, repo, state, opts)
+else:
+# Unreachable code, but help type checkers not think that
+# 'basename' may be used before initialization when checking
+# ShelfDir below.
+raise error.ProgrammingError("neither abort nor continue specified")
 elif len(shelved) > 1:
 raise error.InputError(_(b'can only unshelve one change at a time'))
 elif not shelved:
 shelved = ShelfDir(repo).listshelves()
 if not shelved:
 raise error.InputError(_(b"shelved change '%s' not found") % basename)
 return _dounshelve(ui, repo, basename, opts)
-def _dounshelve(ui, repo, basename, opts):
+def _dounshelve(ui, repo, basename: bytes, opts) -> None:
 repo = repo.unfiltered()
 lock = tr = None
 try:
 lock = repo.lock()
 tr = repo.transaction(b'unshelve', report=lambda x: None)

Mercurial > public > mercurial-scm > hg

comparison mercurial/shelve.py @ 51812:a1a94d488e14