this way we get at relative revisions, error handling, etc.
- don't run environment for history command unless "current" was requested;
running the environment modifies output with logging, might access multiple
dbs, etc., so don't get into it if not needed
- add test suite for commands, start with history output
- document/changelog for history -r feature
- fix sys.stdout writing for py3k
- one more with_statement future not needed
):
script.run_env()
-def history(config, rev_range=""):
+def history(config, rev_range=None):
"""List changeset scripts in chronological order."""
script = ScriptDirectory.from_config(config)
- def display_history(start=None, end=None):
- revs = iter(script.walk_revisions())
- if end:
- # skip to end
- for sc in revs:
- if sc == end:
- if sc.is_head:
- config.print_stdout("")
- config.print_stdout(sc)
- break
- if (start or end) and end == start:
- return
-
- for sc in revs:
+ if rev_range is not None:
+ if ":" not in rev_range:
+ raise util.CommandError(
+ "History range requires [start]:[end], "
+ "[start]:, or :[end]")
+ base, head = rev_range.strip().split(":")
+ else:
+ base = head = None
+
+ def _display_history(config, script, base, head):
+ for sc in script.walk_revisions(
+ base=base or "base",
+ head=head or "head"):
if sc.is_head:
config.print_stdout("")
config.print_stdout(sc)
- if sc == start:
- break
- if not rev_range:
- return display_history()
-
- if ":" not in rev_range:
- raise ValueError("rev_range must be formatted in '[start]:[end]'") # need a right message
-
-
- def display_history_ragne(rev, context):
- _start, _end = rev_range.split(":", 1)
- _start = _start or "base"
- _end = _end or "head"
+ def _display_history_w_current(config, script, base=None, head=None):
+ def _display_current_history(rev, context):
+ if head is None:
+ _display_history(config, script, base, rev)
+ elif base is None:
+ _display_history(config, script, rev, head)
+ return []
- if _start == 'current':
- _start = rev
- if _end == 'current':
- _end = rev
-
- start = script.get_revision(_start)
- end = script.get_revision(_end)
- display_history(start=start, end=end)
- return []
+ with EnvironmentContext(
+ config,
+ script,
+ fn=_display_current_history
+ ):
+ script.run_env()
- with EnvironmentContext(
- config,
- script,
- fn=display_history_ragne
- ):
- script.run_env()
+ if base == "current":
+ _display_history_w_current(config, script, head=head)
+ elif head == "current":
+ _display_history_w_current(config, script, base=base)
+ else:
+ _display_history(config, script, base, head)
def branches(config):
import os
import sys
-from . import command, util, package_dir
+from . import command, util, package_dir, compat
class Config(object):
"""Represent an Alembic configuration.
def print_stdout(self, text, *arg):
"""Render a message to standard out."""
- self.stdout.write((str(text) % arg) + "\n")
+ self.stdout.write((compat.text_type(text) % arg) + "\n")
@util.memoized_property
def file_config(self):
if 'rev_range' in kwargs:
parser.add_argument("-r", "--rev-range",
action="store",
- help="Specify the range of display revisions. "
- "range is formatted in [start]:[end] "
- 'accepting any rev number, "head", "base", '
- 'or "current". '
- 'the left side of : defaults to "base" '
- 'the right side defaults to "head"')
+ help="Specify a revision range; "
+ "format is [start]:[end]")
positional_help = {
"found in configuration.")
return ScriptDirectory(
util.coerce_resource_to_filename(script_location),
- file_template = config.get_main_option(
+ file_template=config.get_main_option(
'file_template',
_default_file_template)
)
- def walk_revisions(self):
+ def walk_revisions(self, base="base", head="head"):
"""Iterate through all revisions.
This is actually a breadth-first tree traversal,
with leaf nodes being heads.
"""
- heads = set(self.get_heads())
- base = self.get_revision("base")
+ if head == "head":
+ heads = set(self.get_heads())
+ else:
+ heads = set([head])
while heads:
todo = set(heads)
heads = set()
.. changelog::
:version: 0.6.0
+ .. change::
+ :tags: feature
+ :pullreq: 35
+
+ Added ``-r`` argument to ``alembic history`` command,
+ allows specification of ``[start]:[end]`` to view
+ a slice of history. Accepts revision numbers, symbols
+ "base", "head", a new symbol "current" representing the
+ current migration, as well as relative ranges for one
+ side at a time (i.e. ``-r-5:head``, ``-rcurrent:+3``).
+ Courtesy Atsushi Odagiri for this feature.
+
.. change::
:tags: feature
:pullreq: 34
Alembic will stop and let you know if more than one version starts with that prefix.
+Viewing History Ranges
+----------------------
+
+Using the ``-r`` option to ``alembic history``, we can also view various slices
+of history. The ``-r`` argument accepts an argument ``[start]:[end]``, where
+either may be a revision number, or various combinations of ``base``, ``head``,
+``currrent`` to specify the current revision, as well as negative relative
+ranges for ``[start]`` and positive relative ranges for ``[end]``::
+
+ $ alembic history -r1975ea:ae1027
+
+A relative range starting from three revs ago up to current migration,
+which will invoke the migration environment against the database
+to get the current migration::
+
+ $ alembic history -r-3:current
+
+View all revisions from 1975 to the head::
+
+ $ alembic history -r1975ea:
+
+.. versionadded:: 0.6.1 ``alembic revision`` now accepts the ``-r`` argument to
+ specify specific ranges based on version numbers, symbols, or relative deltas.
+
+
+
Downgrading
===========
script = ScriptDirectory.from_config(cfg)
script.generate_revision(a, "revision a", refresh=True)
- write_script(script, a, """
+ write_script(script, a, """\
+"Rev A"
revision = '%s'
down_revision = None
""" % a)
script.generate_revision(b, "revision b", refresh=True)
- write_script(script, b, """
+ write_script(script, b, """\
+"Rev B"
revision = '%s'
down_revision = '%s'
""" % (b, a))
script.generate_revision(c, "revision c", refresh=True)
- write_script(script, c, """
+ write_script(script, c, """\
+"Rev C"
revision = '%s'
down_revision = '%s'
--- /dev/null
+import unittest
+from . import clear_staging_env, staging_env, \
+ _sqlite_testing_config, \
+ three_rev_fixture, eq_
+from alembic import command
+from io import StringIO
+
+class StdoutCommandTest(unittest.TestCase):
+
+ @classmethod
+ def setup_class(cls):
+ cls.env = staging_env()
+ cls.cfg = _sqlite_testing_config()
+ cls.a, cls.b, cls.c = three_rev_fixture(cls.cfg)
+
+ @classmethod
+ def teardown_class(cls):
+ clear_staging_env()
+
+ def _eq_cmd_output(self, buf, expected):
+ revs = {"reva": self.a, "revb": self.b, "revc": self.c}
+ eq_(
+ [s for s in buf.getvalue().split("\n") if s],
+ [exp % revs for exp in expected]
+ )
+
+ def test_history_full(self):
+ self.cfg.stdout = buf = StringIO()
+ command.history(self.cfg)
+ self._eq_cmd_output(buf, [
+ '%(revb)s -> %(revc)s (head), Rev C',
+ '%(reva)s -> %(revb)s, Rev B',
+ 'None -> %(reva)s, Rev A'
+ ])
+
+ def test_history_num_range(self):
+ self.cfg.stdout = buf = StringIO()
+ command.history(self.cfg, "%s:%s" % (self.a, self.b))
+ self._eq_cmd_output(buf, [
+ '%(reva)s -> %(revb)s, Rev B',
+ ])
+
+ def test_history_base_to_num(self):
+ self.cfg.stdout = buf = StringIO()
+ command.history(self.cfg, ":%s" % (self.b))
+ self._eq_cmd_output(buf, [
+ '%(reva)s -> %(revb)s, Rev B',
+ 'None -> %(reva)s, Rev A'
+ ])
+
+ def test_history_num_to_head(self):
+ self.cfg.stdout = buf = StringIO()
+ command.history(self.cfg, "%s:" % (self.a))
+ self._eq_cmd_output(buf, [
+ '%(revb)s -> %(revc)s (head), Rev C',
+ '%(reva)s -> %(revb)s, Rev B',
+ ])
+
+ def test_history_num_plus_relative(self):
+ self.cfg.stdout = buf = StringIO()
+ command.history(self.cfg, "%s:+2" % (self.a))
+ self._eq_cmd_output(buf, [
+ '%(revb)s -> %(revc)s (head), Rev C',
+ '%(reva)s -> %(revb)s, Rev B',
+ ])
+
+ def test_history_relative_to_num(self):
+ self.cfg.stdout = buf = StringIO()
+ command.history(self.cfg, "-2:%s" % (self.c))
+ self._eq_cmd_output(buf, [
+ '%(revb)s -> %(revc)s (head), Rev C',
+ '%(reva)s -> %(revb)s, Rev B',
+ ])
+
+ def test_history_current_to_head_as_b(self):
+ command.stamp(self.cfg, self.b)
+ self.cfg.stdout = buf = StringIO()
+ command.history(self.cfg, "current:")
+ self._eq_cmd_output(buf, [
+ '%(revb)s -> %(revc)s (head), Rev C',
+ ])
+
+ def test_history_current_to_head_as_base(self):
+ command.stamp(self.cfg, "base")
+ self.cfg.stdout = buf = StringIO()
+ command.history(self.cfg, "current:")
+ self._eq_cmd_output(buf, [
+ '%(revb)s -> %(revc)s (head), Rev C',
+ '%(reva)s -> %(revb)s, Rev B',
+ 'None -> %(reva)s, Rev A'
+ ])
# coding: utf-8
-from __future__ import with_statement, unicode_literals
+from __future__ import unicode_literals
import unittest