from alembic import options, util
import os
import sys
+import uuid
def list_templates(opts):
"""List available templates"""
util.status("Creating directory %s" % os.path.abspath(dir_),
os.makedirs, dir_)
- script = ScriptDirectory(dir_)
+ script = ScriptDirectory(dir_, opts)
template_dir = os.path.join(opts.get_template_directory(),
opts.cmd_line_options.template)
if not os.access(template_dir, os.F_OK):
opts.err("No such template %r" % opts.cmd_line_options.template)
+
for file_ in os.listdir(template_dir):
if file_ == 'alembic.ini.mako':
config_file = os.path.abspath(opts.cmd_line_options.config)
def revision(opts):
"""Create a new revision file."""
+
+ script = ScriptDirectory.from_options(opts)
+ script.generate_rev(uuid.uuid4())
def upgrade(opts):
"""Upgrade to the latest version."""
import os
from alembic import util
import shutil
+import re
+import inspect
+
+_uuid_re = re.compile(r'[a-z0-9]{16}')
+_mod_def_re = re.compile(r'(upgrade|downgrade)_([a-z0-9]{16})')
class ScriptDirectory(object):
- def __init__(self, dir):
+ def __init__(self, dir, options):
self.dir = dir
+ self.options = otions
@classmethod
def from_options(cls, options):
- return ScriptDirectory(options.get_main_option('script_location'))
+ return ScriptDirectory(
+ options.get_main_option('script_location'),
+ options)
+ @util.memoized_property
+ def _revision_map(self):
+ for file_ in os.listdir(self.dir):
+ script = Script.from_file(self.dir, file_)
+ if script is None:
+ continue
+ map_[script.upgrade] = script
+ return map_
+
+ def _get_head(self):
+ # TODO: keep map sorted chronologically
+
+ for script in self._revision_map.values():
+ if script.upgrade is None \
+ and script.downgrade in self._revision_map:
+ return script
+ else:
+ return None
+
+ def _get_origin(self):
+ # TODO: keep map sorted chronologically
+
+ for script in self._revision_map.values():
+ if script.downgrade is None \
+ and script.upgrade in self._revision_map:
+ return script
+ else:
+ return None
+
def generate_template(self, src, dest, **kw):
util.status("Generating %s" % os.path.abspath(src),
util.template_to_file,
util.status("Generating %s" % os.path.abspath(dest),
shutil.copy,
src, dest)
+
+
+ def generate_rev(self, revid):
+ current_head = self._get_head()
+ self.generate_template(
+ os.path.join(self.dir, "script.py.mako",
+ up_revision=revid,
+ down_revision=current_head.upgrade if current_head else None
+ )
+ )
+
+class Script(object):
+ def __init__(self, module):
+ self.module = module
+ self.upgrade = self.downgrade = None
+ for name in dir(module):
+ m = _mod_def_re.match(name)
+ if not m:
+ continue
+ fn = getattr(module, name)
+ if not inspect.isfunction(fn):
+ continue
+ if m.group(1) == 'upgrade':
+ self.upgrade = m.group(2)
+ elif m.group(1) == 'downgrade':
+ self.downgrade = m.group(2)
+ if not self.downgrade and not self.upgrade:
+ raise Exception("Script %s has no upgrade or downgrade path" % module)
+
+ @classmethod
+ def from_path(cls, dir_, filename):
+ if not _uuid_re.match(filename):
+ return None
+
+ module = util.load_python_file(dir_, filename)
+ return Script(module)
\ No newline at end of file
import os
import textwrap
from sqlalchemy import util
+import imp
NO_VALUE = util.symbol("NO_VALUE")
for line in lines[0:-1]:
sys.stdout.write(" " +line + "\n")
sys.stdout.write(" " + lines[-1] + ("\n" if newline else ""))
+
+def load_python_file(dir_, filename):
+ """Load a file from the given path as a Python module."""
+
+ module_id = re.sub(r'\W', "_", filename)
+ path = os.path.join(dir_, filename)
+ module = imp.load_source(module_id, path, open(path, 'rb'))
+ del sys.modules[module_id]
+ return module
+
+class memoized_property(object):
+ """A read-only @property that is only evaluated once."""
+ def __init__(self, fget, doc=None):
+ self.fget = fget
+ self.__doc__ = doc or fget.__doc__
+ self.__name__ = fget.__name__
+
+ def __get__(self, obj, cls):
+ if obj is None:
+ return None
+ obj.__dict__[self.__name__] = result = self.fget(obj)
+ return result
def upgrade_${up_revision}():
pass
+% if down_revision:
def downgrade_${down_revision}():
pass
+% else:
+# this is the origin node, no downgrade !
+% endif
from alembic.op import *
-% for engine in engines:
-def upgrade_${engine}_${up_revision}():
- pass
+def upgrade_${up_revision}(engine):
+ eval("upgrade_%s_${up_revision}" % engine.name)()
-def downgrade_${engine}_${down_revision}():
- pass
+% if down_revision:
+def downgrade_${down_revision}(engine):
+ eval("upgrade_%s_${down_revision}" % engine.name)()
+% else:
+# this is the origin node, no downgrade !
+% endif
+
+
+% for engine in ["engine1", "engine2"]:
+ def upgrade_${engine}_${up_revision}():
+ pass
+
+ def downgrade_${engine}_${down_revision}():
+ pass
% endfor
\ No newline at end of file
def upgrade_${up_revision}():
pass
+% if down_revision:
def downgrade_${down_revision}():
pass
+% else:
+# this is the origin node, no downgrade !
+% endif
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
