Mercurial > hg > ltpdarepo
changeset 147:85b2bc3c7e04
Add database schema dump utility.
| author | Daniele Nicolodi <daniele@grinta.net> |
|---|---|
| date | Fri, 21 Oct 2011 14:47:30 +0200 |
| parents | 7c5bd2a8bb3f |
| children | f594c49a3da4 |
| files | develop.cfg src/ltpdarepo/tests/dump.py src/ltpdarepo/tests/test_schema.py src/setup.py |
| diffstat | 4 files changed, 110 insertions(+), 100 deletions(-) [+] |
line wrap: on
line diff
--- a/develop.cfg Fri Oct 21 16:10:17 2011 +0200 +++ b/develop.cfg Fri Oct 21 14:47:30 2011 +0200 @@ -8,7 +8,7 @@ coverage zope.testbrowser [wsgi] SQLAlchemy -scripts += unit2 coverage +scripts += unit2 coverage dump [omelette] recipe = collective.recipe.omelette
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/src/ltpdarepo/tests/dump.py Fri Oct 21 14:47:30 2011 +0200 @@ -0,0 +1,102 @@ +from contextlib import contextmanager +from cStringIO import StringIO + +import sqlalchemy + + +class Formatter(object): + def __init__(self, out): + self.level = 0 + self.out = out + + @contextmanager + def indent(self, increase=1): + self.level += increase + yield + self.level -= increase + + def write(self, string): + self.out.write(' ' * self.level) + self.out.write(string) + self.out.write('\n') + + +def dump_schema(username, password, hostname, database, out): + f = Formatter(out) + url = sqlalchemy.engine.url.URL('mysql', username=username, password=password, host=hostname, database=database) + engine = sqlalchemy.engine.create_engine(url) + inspector = sqlalchemy.engine.reflection.Inspector.from_engine(engine) + dump_database(inspector, f) + + +def dump_database(inspector, f): + f.write(inspector.default_schema_name) + with f.indent(): + # tables + for table in inspector.get_table_names(): + f.write('table: %s' % table) + with f.indent(): + dump_table(inspector, table, f) + # views + for view in inspector.get_view_names(): + f.write('view: %s' % view) + with f.indent(): + dump_table(inspector, view, f) + + +def dump_table(inspector, table, f): + # columns + f.write('columns:') + with f.indent(): + for column in inspector.get_columns(table): + f.write('- %s' % column.pop('name')) + with f.indent(): + for k, v in column.iteritems(): + f.write('%s: %s' % (k, v)) + # primary keys + pks = inspector.get_primary_keys(table) + if pks: + f.write('primary keys:') + with f.indent(): + for pk in pks: + f.write('- %s' % pk) + # foreign keys + fks = inspector.get_foreign_keys(table) + if fks: + f.write('foreign keys:') + with f.indent(): + for fk in fks: + f.write('- %s' % fk.pop('name')) + fk.pop('options') + with f.indent(): + for k, v in fk.iteritems(): + if isinstance(v, list): + v = ', '.join(map(unicode, v)) + f.write('%s: %s' % (k, v)) + # indexes + indexes = inspector.get_indexes(table) + if indexes: + f.write('indexes:') + with f.indent(): + for index in indexes: + f.write('- %s' % index.pop('name')) + with f.indent(): + for k, v in index.iteritems(): + if isinstance(v, list): + v = ', '.join(map(unicode, v)) + f.write('%s: %s' % (k, v)) + # options + options = inspector.get_table_options(table) + if options: + f.write('options:') + with f.indent(): + for k, v in options.iteritems(): + if k.startswith('mysql_'): + k = k[6:] + f.write('- %s: %s' % (k, v)) + + +def main(): + import sys + username, password, hostname, database = sys.argv[1:] + dump_schema(username, password, hostname, database, sys.stdout)
--- a/src/ltpdarepo/tests/test_schema.py Fri Oct 21 16:10:17 2011 +0200 +++ b/src/ltpdarepo/tests/test_schema.py Fri Oct 21 14:47:30 2011 +0200 @@ -4,106 +4,13 @@ import os import warnings -from contextlib import contextmanager from cStringIO import StringIO import MySQLdb as mysql -import sqlalchemy from ltpdarepo.config import USERNAME, PASSWORD, HOSTNAME from ltpdarepo.admin import wipe, install, createdb, useradd, grant, upgrade - - -class Formatter(object): - def __init__(self, out=sys.stdout): - self.level = 0 - self.out = out - - @contextmanager - def indent(self, increase=1): - self.level += increase - yield - self.level -= increase - - def write(self, string): - self.out.write(' ' * self.level) - self.out.write(string) - self.out.write('\n') - - -def dump_schema(database, out): - f = Formatter(out) - url = sqlalchemy.engine.url.URL('mysql', username=USERNAME, password=PASSWORD, host=HOSTNAME, database=database) - engine = sqlalchemy.engine.create_engine(url) - inspector = sqlalchemy.engine.reflection.Inspector.from_engine(engine) - dump_database(inspector, f) - - -def dump_database(inspector, f): - f.write(inspector.default_schema_name) - with f.indent(): - # tables - for table in inspector.get_table_names(): - f.write('table: %s' % table) - with f.indent(): - dump_table(inspector, table, f) - # views - for view in inspector.get_view_names(): - f.write('view: %s' % view) - with f.indent(): - dump_table(inspector, view, f) - - -def dump_table(inspector, table, f): - # columns - f.write('columns:') - with f.indent(): - for column in inspector.get_columns(table): - f.write('- %s' % column.pop('name')) - with f.indent(): - for k, v in column.iteritems(): - f.write('%s: %s' % (k, v)) - # primary keys - pks = inspector.get_primary_keys(table) - if pks: - f.write('primary keys:') - with f.indent(): - for pk in pks: - f.write('- %s' % pk) - # foreign keys - fks = inspector.get_foreign_keys(table) - if fks: - f.write('foreign keys:') - with f.indent(): - for fk in fks: - f.write('- %s' % fk.pop('name')) - fk.pop('options') - with f.indent(): - for k, v in fk.iteritems(): - if isinstance(v, list): - v = ', '.join(map(unicode, v)) - f.write('%s: %s' % (k, v)) - # indexes - indexes = inspector.get_indexes(table) - if indexes: - f.write('indexes:') - with f.indent(): - for index in indexes: - f.write('- %s' % index.pop('name')) - with f.indent(): - for k, v in index.iteritems(): - if isinstance(v, list): - v = ', '.join(map(unicode, v)) - f.write('%s: %s' % (k, v)) - # options - options = inspector.get_table_options(table) - if options: - f.write('options:') - with f.indent(): - for k, v in options.iteritems(): - if k.startswith('mysql_'): - k = k[6:] - f.write('- %s: %s' % (k, v)) +from ltpdarepo.tests.dump import dump_schema class TestCase(unittest.TestCase): @@ -148,8 +55,8 @@ # dump database structure upgraded = StringIO() - dump_schema('ltpda', upgraded) - dump_schema('db1', upgraded) + dump_schema(USERNAME, PASSWORD, HOSTNAME, 'ltpda', upgraded) + dump_schema(USERNAME, PASSWORD, HOSTNAME, 'db1', upgraded) upgraded.seek(0) # install @@ -162,8 +69,8 @@ # dump database structure new = StringIO() - dump_schema('ltpda', new) - dump_schema('db1', new) + dump_schema(USERNAME, PASSWORD, HOSTNAME, 'ltpda', new) + dump_schema(USERNAME, PASSWORD, HOSTNAME, 'db1', new) new.seek(0) # compare
--- a/src/setup.py Fri Oct 21 16:10:17 2011 +0200 +++ b/src/setup.py Fri Oct 21 14:47:30 2011 +0200 @@ -6,5 +6,6 @@ name='ltpdarepo', version=VERSION, entry_points={'console_scripts': ['run = ltpdarepo:main', - 'admin = ltpdarepo.admin:main']} + 'admin = ltpdarepo.admin:main', + 'dump = ltpdarepo.tests.dump:main']} )