Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if check:
per_target = set()
while todo:
rev = todo.pop()
if check:
per_target.add(rev)
if rev in seen:
continue
seen.add(rev)
todo.extend(
map_[rev_id] for rev_id in fn(rev))
yield rev
if check and per_target.intersection(targets).difference([target]):
raise RevisionError(
"Requested revision %s overlaps with "
"other requested revisions" % target.revision)
"argument '%(head_arg)s'; please "
"specify a specific target revision, "
"'@%(head_arg)s' to "
"narrow to a specific head, or 'heads' for all heads")
multiple_heads = multiple_heads % {
"head_arg": end or mh.argument,
"heads": str(mh.heads)
}
raise Exception(multiple_heads)
except ResolutionError as re:
if resolution is None:
resolution = "Can't locate revision identified by '%s'" % (
re.argument
)
raise Exception(resolution)
except RevisionError as err:
raise Exception(err.args[0])
"Revision %s is not an ancestor of revision %s" %
(lower or "base", upper or "base")
)
class MultipleHeads(RevisionError):
def __init__(self, heads, argument):
self.heads = heads
self.argument = argument
super(MultipleHeads, self).__init__(
"Multiple heads are present for given argument '%s'; "
"%s" % (argument, ", ".join(heads))
)
class ResolutionError(RevisionError):
def __init__(self, message, argument):
super(ResolutionError, self).__init__(message)
self.argument = argument
"""
emmett.orm.migrations.exceptions
--------------------------------
Provides exceptions for migration operations.
:copyright: (c) 2014-2019 by Giovanni Barillari
:license: BSD, see LICENSE for more details.
"""
class RevisionError(Exception):
pass
class RangeNotAncestorError(RevisionError):
def __init__(self, lower, upper):
self.lower = lower
self.upper = upper
super(RangeNotAncestorError, self).__init__(
"Revision %s is not an ancestor of revision %s" %
(lower or "base", upper or "base")
)
class MultipleHeads(RevisionError):
def __init__(self, heads, argument):
self.heads = heads
self.argument = argument
super(MultipleHeads, self).__init__(
"Multiple heads are present for given argument '%s'; "
"%s" % (argument, ", ".join(heads))
class RevisionError(Exception):
pass
class RangeNotAncestorError(RevisionError):
def __init__(self, lower, upper):
self.lower = lower
self.upper = upper
super(RangeNotAncestorError, self).__init__(
"Revision %s is not an ancestor of revision %s" %
(lower or "base", upper or "base")
)
class MultipleHeads(RevisionError):
def __init__(self, heads, argument):
self.heads = heads
self.argument = argument
super(MultipleHeads, self).__init__(
"Multiple heads are present for given argument '%s'; "
"%s" % (argument, ", ".join(heads))
)
class ResolutionError(RevisionError):
def __init__(self, message, argument):
super(ResolutionError, self).__init__(message)
self.argument = argument
len(total_space.intersection(rev._all_nextrev)) > 1
)
# it's not possible for any "uppers" to be in branch_todo,
# because the ._all_nextrev of those nodes is not in total_space
#assert not branch_todo.intersection(uppers)
todo = deque(
r for r in uppers if r.revision in total_space)
# iterate for total_space being emptied out
total_space_modified = True
while total_space:
if not total_space_modified:
raise RevisionError(
"Dependency resolution failed; iteration can't proceed")
total_space_modified = False
# when everything non-branch pending is consumed,
# add to the todo any branch nodes that have no
# descendants left in the queue
if not todo:
todo.extendleft(
sorted(
(
rev for rev in branch_todo
if not rev._all_nextrev.intersection(total_space)
),
# favor "revisioned" branch points before
# dependent ones
key=lambda rev: 0 if rev.is_branch_point else 1
)