Skip to content
Find file
Fetching contributors…
Cannot retrieve contributors at this time
executable file 296 lines (227 sloc) 9.33 KB
#!/usr/bin/env python
The worst schema versioning system, ever?
Usage: path/to/schema_files
schematic talks to your system over stdin on the command line. It will do
SQL migrations by talking directly to your database. It will also do
migrations for any language, so you can do almost anything in the migration.
It supports all DBMSs that have a command line interface and doesn't
care what programming language you worship. Win!
Schematic expects 1 argument which is the directory full of schema, DDL or
script files you wish to migrate.
Configuration is done in ``, which should look something like:
# How to connect to the database
db = 'mysql --silent -p blam -D pow'
# The table where version info is stored.
table = 'schema_version'
# Optionally, how you want to handle something that's not SQL
handlers = {'.py': 'python -B runscript migrations.%s'}
It's python so you can do whatever crazy things you want, and it's a
separate file so you can keep local settings out of version control.
schematic will try to look for on the PYTHON_PATH and then
in the migrations directory.
Migrations are just files whose names start with a number, like
`001-adding-awesome.sql`. They're matched against `'^\d+'` so you can
put zeros in front to keep ordering in `ls` happy, and whatever you want
after the migration number, such as text describing the migration.
If a file ends with no extension or .sql it is treated as SQL and passed
to your database. If the file ends with an extension in handlers, the file
is passed to that command to do with as it wishes.
schematic creates a table (named in with one column, that
holds one row, which describes the current version of the database. Any
migration file with a number greater than the current version will be
applied to the database and the version tracker will be upgraded. The
migration and version bump are performed in a transaction.
The version-tracking table will initially be set to 0, so the 0th
migration could be a script that creates all your tables (for
reference). Migration numbers are not required to increase linearly.
schematic doesn't pretend to be intelligent. Running migrations manually
without upgrading the version tracking will throw things off.
Tested on sqlite and mysql.
NOTE: any superfluous output, like column headers, will cause an error.
On mysql, this is fixed by using the `--silent` parameter.
Things that might be nice: downgrades, coping with missing migrations.
import optparse
import os
import re
import sys
import time
from subprocess import Popen, PIPE, STDOUT
SETTINGS = 'settings'
VARIABLES = ['db', 'table', 'handlers']
OPTIONAL_VARIABLES = {'handlers': {}}
IGNORE_FILES = ['.pyc']
COUNT = 'SELECT COUNT(version) FROM %s;'
SELECT = 'SELECT version FROM %s;'
INSERT = 'INSERT INTO %s (version) VALUES (%s);'
UPDATE = 'UPDATE %s SET version = %s;'
class SchematicError(Exception):
"""Base class for custom errors."""
def exception(f):
"""Decorator to turn a function into an subclass of SchematicError."""
class E(SchematicError):
def __init__(self, *args, **kwargs):
msg = f(self, *args, **kwargs)
SchematicError.__init__(self, msg)
E.__name__ = f.__name__
return E
def MissingSettings(self):
return "Couldn't import settings file"
def SettingsError(self, k):
return "Couldn't find value for '%s' in" % (k, SETTINGS)
def DbError(self, cmd, stdout, stderr, returncode):
msg = '\n'.join(["Had trouble running this: %s", "stdout: %s",
"stderr: %s", "returncode: %s"])
return msg % (cmd, stdout, stderr, returncode)
def MultipleMigrations(self, num):
return 'Multiple migrations with number: %d' % num
def ExternalError(self, cmd, stdout, stderr, returncode):
return DbError(cmd, stdout, stderr, returncode)
def get_settings(schema_dir):
# Also search for settings in the schema_dir.
import schematic_settings as settings
except ImportError, e:
import settings
except ImportError, e:
raise MissingSettings
for k in VARIABLES:
getattr(settings, k)
except AttributeError:
raise SettingsError(k)
return settings
def say(db, command):
"""Try talking to the database, bail if there's anything on stderr
or a bad returncode."""
p = Popen(db, stdin=PIPE, stdout=PIPE, stderr=PIPE, shell=True)
stdout, stderr = p.communicate(command)
if stderr or p.returncode != 0:
raise DbError(command, stdout, stderr, p.returncode)
return stdout
def ext(command):
"""Run an external command and blow up if theres an error."""
p = Popen(command, stdin=PIPE, stdout=PIPE, stderr=STDOUT, shell=True)
stdout, stderr = p.communicate()
# Ignoring stderr because we don't want to break on deprecation warnings
# if you want to stop the migration, raise an error.
if p.returncode != 0:
raise ExternalError(command, stdout, stderr, p.returncode)
return stdout
def table_check(db, table):
# Try a count to see if the table is there.
count = int(say(db, COUNT % table))
except DbError:
# Try to create the table.
say(db, CREATE % table)
count = 0
# Start tracking at version 0.
if count == 0:
say(db, INSERT % (table, 0))
def find_upgrades(schema_dir):
fullpath = lambda p: os.path.join(schema_dir, p)
files = filter(os.path.isfile,
map(fullpath, os.listdir(schema_dir)))
upgrades = {}
for f in files:
if os.path.splitext(f)[1] in IGNORE_FILES:
m = re.match('^(\d+)', os.path.basename(f))
if m:
num = int(
if num in upgrades:
raise MultipleMigrations(num)
upgrades[num] = f
return upgrades
def run_upgrades(db, table, schema_dir, maximum=None, handlers={}):
current = get_version(db, table)
all_upgrades = find_upgrades(schema_dir).items()
upgrades = [(version, path) for version, path in all_upgrades
if version > current]
for version, path in sorted(upgrades):
if maximum and version > maximum:
print 'Reached max version: %s' % maximum
start = time.time()
upgrade(db, table, version, path, handlers)
print 'That took %.2f seconds' % (time.time() - start)
print '#' * 50, '\n'
def upgrade(db, table, version, path, handlers):
extension = os.path.splitext(path)[1]
handler = handlers.get(extension)
update = UPDATE % (table, version)
if not extension or extension == '.sql':
sql = open(path).read()
print 'Running SQL migration %s:\n' % version, sql
say(db, UPGRADE % (sql, update))
elif handler:
cmd = handler % (os.path.splitext(os.path.basename(path))[0])
print 'Running %s migation %s:\n' % (extension, version), cmd
print ext(cmd)
say(db, update)
raise NotImplementedError("Don't know how to migrate: %s" % path)
def get_version(db, table):
return int(say(db, SELECT % table))
def main(schema_dir, maximum=None):
settings = get_settings(schema_dir)
db, table = settings.db, settings.table
table_check(db, table)
if maximum:
print 'Up to max: %s' % maximum
run_upgrades(db, table, schema_dir, maximum,
getattr(settings, 'handlers',
def update(schema_dir, version):
settings = get_settings(schema_dir)
table_check(settings.db, settings.table)
say(settings.db, UPDATE % (settings.table, version))
def version(schema_dir):
settings = get_settings(schema_dir)
print get_version(settings.db, settings.table)
if __name__ == '__main__':
d = '/path/to/migrations/dir'
error = "Expected a directory: %s" % d
# No arguments yet, but we'll get there.
parser = optparse.OptionParser(usage="Usage: %%prog %s" % d)
parser.add_option('-u', '--update', dest='update', default=False,
help='Update schema tracking table to this version '
'(without running any migrations)')
parser.add_option('-v', '--version', dest='version', default=False,
help='Print the current schema version (without '
'running any migrations)')
parser.add_option('-m', '--max', dest='max', default=None,
action='store', type='int',
help='Stop running migrations after the specified '
options, args = parser.parse_args()
if len(args) != 1:
path = os.path.realpath(args[0])
if not os.path.isdir(path):
if options.update:
update(path, options.update)
elif options.version:
main(path, options.max)
except SchematicError, e:
print 'Error:', e
Jump to Line
Something went wrong with that request. Please try again.