( )⚙ D10126 typing: add some type annotations to mercurial/phases.py

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D10126

typing: add some type annotations to mercurial/phases.py
ClosedPublic

Authored by mharbison72 on Mar 6 2021, 7:46 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG77e129be10de: typing: add some type annotations to mercurial/phases.py
Summary

Some of these were helpful in typing other modules, and then I typed the
easy-ish ones. Black forces the long Phasedefaults definition to be wrapped,
which pytype seems OK with (as shown with reveal_type()), but it does seem to
confuse PyCharm a bit.

Diff Detail

Repository
rHG Mercurial
Branch
default
Lint
No Linters Available
Unit
No Unit Test Coverage

Event Timeline

mharbison72 created this revision.Mar 6 2021, 7:46 PM
mharbison72 added a commit: rHG77e129be10de: typing: add some type annotations to mercurial/phases.py.Mar 8 2021, 3:25 PM
This revision was not accepted when it landed; it landed in state Needs Review.
Closed by commit rHG77e129be10de: typing: add some type annotations to mercurial/phases.py (authored by mharbison72). · Explain Why
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/phases.py (41 lines)

Diff 26123

mercurial/phases.py

error, error,
pycompat, pycompat,
requirements, requirements,
smartset, smartset,
txnutil, txnutil,
util, util,
) )
if pycompat.TYPE_CHECKING:
from typing import (
Any,
Callable,
Dict,
Iterable,
List,
Optional,
Set,
Tuple,
)
from . import (
localrepo,
ui as uimod,
)
Phaseroots = Dict[int, Set[bytes]]
Phasedefaults = List[
Callable[[localrepo.localrepository, Phaseroots], Phaseroots]
]
_fphasesentry = struct.Struct(b'>i20s') _fphasesentry = struct.Struct(b'>i20s')
# record phase index # record phase index
public, draft, secret = range(3) public, draft, secret = range(3) # type: int
archived = 32 # non-continuous for compatibility archived = 32 # non-continuous for compatibility
internal = 96 # non-continuous for compatibility internal = 96 # non-continuous for compatibility
allphases = (public, draft, secret, archived, internal) allphases = (public, draft, secret, archived, internal)
trackedphases = (draft, secret, archived, internal) trackedphases = (draft, secret, archived, internal)
# record phase names # record phase names
cmdphasenames = [b'public', b'draft', b'secret'] # known to `hg phase` command cmdphasenames = [b'public', b'draft', b'secret'] # known to `hg phase` command
phasenames = dict(enumerate(cmdphasenames)) phasenames = dict(enumerate(cmdphasenames))
phasenames[archived] = b'archived' phasenames[archived] = b'archived'
phasenames[internal] = b'internal' phasenames[internal] = b'internal'
# map phase name to phase number # map phase name to phase number
phasenumber = {name: phase for phase, name in phasenames.items()} phasenumber = {name: phase for phase, name in phasenames.items()}
# like phasenumber, but also include maps for the numeric and binary # like phasenumber, but also include maps for the numeric and binary
# phase number to the phase number # phase number to the phase number
phasenumber2 = phasenumber.copy() phasenumber2 = phasenumber.copy()
phasenumber2.update({phase: phase for phase in phasenames}) phasenumber2.update({phase: phase for phase in phasenames})
phasenumber2.update({b'%i' % phase: phase for phase in phasenames}) phasenumber2.update({b'%i' % phase: phase for phase in phasenames})
# record phase property # record phase property
mutablephases = (draft, secret, archived, internal) mutablephases = (draft, secret, archived, internal)
remotehiddenphases = (secret, archived, internal) remotehiddenphases = (secret, archived, internal)
localhiddenphases = (internal, archived) localhiddenphases = (internal, archived)
def supportinternal(repo): def supportinternal(repo):
# type: (localrepo.localrepository) -> bool
"""True if the internal phase can be used on a repository""" """True if the internal phase can be used on a repository"""
return requirements.INTERNAL_PHASE_REQUIREMENT in repo.requirements return requirements.INTERNAL_PHASE_REQUIREMENT in repo.requirements
def _readroots(repo, phasedefaults=None): def _readroots(repo, phasedefaults=None):
# type: (localrepo.localrepository, Optional[Phasedefaults]) -> Tuple[Phaseroots, bool]
"""Read phase roots from disk """Read phase roots from disk
phasedefaults is a list of fn(repo, roots) callable, which are phasedefaults is a list of fn(repo, roots) callable, which are
executed if the phase roots file does not exist. When phases are executed if the phase roots file does not exist. When phases are
being initialized on an existing repository, this could be used to being initialized on an existing repository, this could be used to
set selected changesets phase to something else than public. set selected changesets phase to something else than public.
Return (roots, dirty) where dirty is true if roots differ from Return (roots, dirty) where dirty is true if roots differ from
if phasedefaults: if phasedefaults:
for f in phasedefaults: for f in phasedefaults:
roots = f(repo, roots) roots = f(repo, roots)
dirty = True dirty = True
return roots, dirty return roots, dirty
def binaryencode(phasemapping): def binaryencode(phasemapping):
# type: (Dict[int, List[bytes]]) -> bytes
"""encode a 'phase -> nodes' mapping into a binary stream """encode a 'phase -> nodes' mapping into a binary stream
The revision lists are encoded as (phase, root) pairs. The revision lists are encoded as (phase, root) pairs.
""" """
binarydata = [] binarydata = []
for phase, nodes in pycompat.iteritems(phasemapping): for phase, nodes in pycompat.iteritems(phasemapping):
for head in nodes: for head in nodes:
binarydata.append(_fphasesentry.pack(phase, head)) binarydata.append(_fphasesentry.pack(phase, head))
return b''.join(binarydata) return b''.join(binarydata)
def binarydecode(stream): def binarydecode(stream):
# type: (...) -> Dict[int, List[bytes]]
"""decode a binary stream into a 'phase -> nodes' mapping """decode a binary stream into a 'phase -> nodes' mapping
The (phase, root) pairs are turned back into a dictionary with The (phase, root) pairs are turned back into a dictionary with
the phase as index and the aggregated roots of that phase as value.""" the phase as index and the aggregated roots of that phase as value."""
headsbyphase = {i: [] for i in allphases} headsbyphase = {i: [] for i in allphases}
entrysize = _fphasesentry.size entrysize = _fphasesentry.size
while True: while True:
entry = stream.read(entrysize) entry = stream.read(entrysize)
if r1[0] > rev: if r1[0] > rev:
data.insert(low, (pycompat.xrange(rev, rev + 1), t)) data.insert(low, (pycompat.xrange(rev, rev + 1), t))
else: else:
data.insert(low + 1, (pycompat.xrange(rev, rev + 1), t)) data.insert(low + 1, (pycompat.xrange(rev, rev + 1), t))
class phasecache(object): class phasecache(object):
def __init__(self, repo, phasedefaults, _load=True): def __init__(self, repo, phasedefaults, _load=True):
# type: (localrepo.localrepository, Optional[Phasedefaults], bool) -> None
if _load: if _load:
# Cheap trick to allow shallow-copy without copy module # Cheap trick to allow shallow-copy without copy module
self.phaseroots, self.dirty = _readroots(repo, phasedefaults) self.phaseroots, self.dirty = _readroots(repo, phasedefaults)
self._loadedrevslen = 0 self._loadedrevslen = 0
self._phasesets = None self._phasesets = None
self.filterunknown(repo) self.filterunknown(repo)
self.opener = repo.svfs self.opener = repo.svfs
def hasnonpublicphases(self, repo): def hasnonpublicphases(self, repo):
# type: (localrepo.localrepository) -> bool
"""detect if there are revisions with non-public phase""" """detect if there are revisions with non-public phase"""
repo = repo.unfiltered() repo = repo.unfiltered()
cl = repo.changelog cl = repo.changelog
if len(cl) >= self._loadedrevslen: if len(cl) >= self._loadedrevslen:
self.invalidate() self.invalidate()
self.loadphaserevs(repo) self.loadphaserevs(repo)
return any( return any(
revs revs
for phase, revs in pycompat.iteritems(self.phaseroots) for phase, revs in pycompat.iteritems(self.phaseroots)
if phase != public if phase != public
) )
def nonpublicphaseroots(self, repo): def nonpublicphaseroots(self, repo):
# type: (localrepo.localrepository) -> Set[bytes]
"""returns the roots of all non-public phases """returns the roots of all non-public phases
The roots are not minimized, so if the secret revisions are The roots are not minimized, so if the secret revisions are
descendants of draft revisions, their roots will still be present. descendants of draft revisions, their roots will still be present.
""" """
repo = repo.unfiltered() repo = repo.unfiltered()
cl = repo.changelog cl = repo.changelog
if len(cl) >= self._loadedrevslen: if len(cl) >= self._loadedrevslen:
self.invalidate() self.invalidate()
self.loadphaserevs(repo) self.loadphaserevs(repo)
return set().union( return set().union(
*[ *[
revs revs
for phase, revs in pycompat.iteritems(self.phaseroots) for phase, revs in pycompat.iteritems(self.phaseroots)
if phase != public if phase != public
] ]
) )
def getrevset(self, repo, phases, subset=None): def getrevset(self, repo, phases, subset=None):
# type: (localrepo.localrepository, Iterable[int], Optional[Any]) -> Any
# TODO: finish typing this
"""return a smartset for the given phases""" """return a smartset for the given phases"""
self.loadphaserevs(repo) # ensure phase's sets are loaded self.loadphaserevs(repo) # ensure phase's sets are loaded
phases = set(phases) phases = set(phases)
publicphase = public in phases publicphase = public in phases
if publicphase: if publicphase:
# In this case, phases keeps all the *other* phases. # In this case, phases keeps all the *other* phases.
phases = set(allphases).difference(phases) phases = set(allphases).difference(phases)
for root in roots: for root in roots:
ps.add(root) ps.add(root)
ps.difference_update(lowerroots) ps.difference_update(lowerroots)
lowerroots.update(ps) lowerroots.update(ps)
self._phasesets[phase] = ps self._phasesets[phase] = ps
self._loadedrevslen = len(cl) self._loadedrevslen = len(cl)
def loadphaserevs(self, repo): def loadphaserevs(self, repo):
# type: (localrepo.localrepository) -> None
"""ensure phase information is loaded in the object""" """ensure phase information is loaded in the object"""
if self._phasesets is None: if self._phasesets is None:
try: try:
res = self._getphaserevsnative(repo) res = self._getphaserevsnative(repo)
self._loadedrevslen, self._phasesets = res self._loadedrevslen, self._phasesets = res
except AttributeError: except AttributeError:
self._computephaserevspure(repo) self._computephaserevspure(repo)
def invalidate(self): def invalidate(self):
self._loadedrevslen = 0 self._loadedrevslen = 0
self._phasesets = None self._phasesets = None
def phase(self, repo, rev): def phase(self, repo, rev):
# type: (localrepo.localrepository, int) -> int
# We need a repo argument here to be able to build _phasesets # We need a repo argument here to be able to build _phasesets
# if necessary. The repository instance is not stored in # if necessary. The repository instance is not stored in
# phasecache to avoid reference cycles. The changelog instance # phasecache to avoid reference cycles. The changelog instance
# is not stored because it is a filecache() property and can # is not stored because it is a filecache() property and can
# be replaced without us being notified. # be replaced without us being notified.
if rev == nullrev: if rev == nullrev:
return public return public
if rev < nullrev: if rev < nullrev:
if finalroots != oldroots: if finalroots != oldroots:
self._updateroots( self._updateroots(
targetphase, {tonode(rev) for rev in finalroots}, tr targetphase, {tonode(rev) for rev in finalroots}, tr
) )
return True return True
return False return False
def filterunknown(self, repo): def filterunknown(self, repo):
# type: (localrepo.localrepository) -> None
"""remove unknown nodes from the phase boundary """remove unknown nodes from the phase boundary
Nothing is lost as unknown nodes only hold data for their descendants. Nothing is lost as unknown nodes only hold data for their descendants.
""" """
filtered = False filtered = False
has_node = repo.changelog.index.has_node # to filter unknown nodes has_node = repo.changelog.index.has_node # to filter unknown nodes
for phase, nodes in pycompat.iteritems(self.phaseroots): for phase, nodes in pycompat.iteritems(self.phaseroots):
missing = sorted(node for node in nodes if not has_node(node)) missing = sorted(node for node in nodes if not has_node(node))
set new changeset in their target phase (or higher). set new changeset in their target phase (or higher).
""" """
phcache = repo._phasecache.copy() phcache = repo._phasecache.copy()
phcache.registernew(repo, tr, targetphase, revs) phcache.registernew(repo, tr, targetphase, revs)
repo._phasecache.replace(phcache) repo._phasecache.replace(phcache)
def listphases(repo): def listphases(repo):
# type: (localrepo.localrepository) -> Dict[bytes, bytes]
"""List phases root for serialization over pushkey""" """List phases root for serialization over pushkey"""
# Use ordered dictionary so behavior is deterministic. # Use ordered dictionary so behavior is deterministic.
keys = util.sortdict() keys = util.sortdict()
value = b'%i' % draft value = b'%i' % draft
cl = repo.unfiltered().changelog cl = repo.unfiltered().changelog
for root in repo._phasecache.phaseroots[draft]: for root in repo._phasecache.phaseroots[draft]:
if repo._phasecache.phase(repo, cl.rev(root)) <= draft: if repo._phasecache.phase(repo, cl.rev(root)) <= draft:
keys[hex(root)] = value keys[hex(root)] = value
# #
# The server can't handle it on it's own as it has no idea of # The server can't handle it on it's own as it has no idea of
# client phase data. # client phase data.
keys[b'publishing'] = b'True' keys[b'publishing'] = b'True'
return keys return keys
def pushphase(repo, nhex, oldphasestr, newphasestr): def pushphase(repo, nhex, oldphasestr, newphasestr):
# type: (localrepo.localrepository, bytes, bytes, bytes) -> bool
"""List phases root for serialization over pushkey""" """List phases root for serialization over pushkey"""
repo = repo.unfiltered() repo = repo.unfiltered()
with repo.lock(): with repo.lock():
currentphase = repo[nhex].phase() currentphase = repo[nhex].phase()
newphase = abs(int(newphasestr)) # let's avoid negative index surprise newphase = abs(int(newphasestr)) # let's avoid negative index surprise
oldphase = abs(int(oldphasestr)) # let's avoid negative index surprise oldphase = abs(int(oldphasestr)) # let's avoid negative index surprise
if currentphase == oldphase and newphase < oldphase: if currentphase == oldphase and newphase < oldphase:
with repo.transaction(b'pushkey-phase') as tr: with repo.transaction(b'pushkey-phase') as tr:
prunestart = repo.revs(b"parents(%ld) and not null", new_heads) prunestart = repo.revs(b"parents(%ld) and not null", new_heads)
pruned = dagop.reachableroots(repo, candidates, prunestart) pruned = dagop.reachableroots(repo, candidates, prunestart)
new_heads.difference_update(pruned) new_heads.difference_update(pruned)
return pycompat.maplist(cl.node, sorted(new_heads)) return pycompat.maplist(cl.node, sorted(new_heads))
def newcommitphase(ui): def newcommitphase(ui):
# type: (uimod.ui) -> int
"""helper to get the target phase of new commit """helper to get the target phase of new commit
Handle all possible values for the phases.new-commit options. Handle all possible values for the phases.new-commit options.
""" """
v = ui.config(b'phases', b'new-commit') v = ui.config(b'phases', b'new-commit')
try: try:
return phasenumber2[v] return phasenumber2[v]
except KeyError: except KeyError:
raise error.ConfigError( raise error.ConfigError(
_(b"phases.new-commit: not a valid phase name ('%s')") % v _(b"phases.new-commit: not a valid phase name ('%s')") % v
) )
def hassecret(repo): def hassecret(repo):
# type: (localrepo.localrepository) -> bool
"""utility function that check if a repo have any secret changeset.""" """utility function that check if a repo have any secret changeset."""
return bool(repo._phasecache.phaseroots[secret]) return bool(repo._phasecache.phaseroots[secret])
def preparehookargs(node, old, new): def preparehookargs(node, old, new):
# type: (bytes, Optional[int], Optional[int]) -> Dict[bytes, bytes]
if old is None: if old is None:
old = b'' old = b''
else: else:
old = phasenames[old] old = phasenames[old]
return {b'node': node, b'oldphase': old, b'phase': phasenames[new]} return {b'node': node, b'oldphase': old, b'phase': phasenames[new]}