Mercurial > hg > ltpdarepo
view src/ltpdarepo/tests/dump.py @ 148:f594c49a3da4
Make schema dump predictable ordering tables and columns by name.
author | Daniele Nicolodi <nicoldi@science.unitn.it> |
---|---|
date | Fri, 21 Oct 2011 19:04:34 +0200 |
parents | 85b2bc3c7e04 |
children |
line wrap: on
line source
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 sorted(inspector.get_table_names()): f.write('table: %s' % table) with f.indent(): dump_table(inspector, table, f) # views for view in sorted(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 sorted(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)