ActiveState Code

#!/usr/bin/python
# -*- coding: cp1250 -*-
__version__ = '$Id: schema_ora.py 525 2008-10-13 08:13:13Z mn $'

# export Oracle schema to text
# usable to compare databases that should be the same
#
# Oracle schema info:
# http://www.eveandersson.com/writing/data-model-reverse-engineering
#
#
# author: Michal Niklas

USAGE = 'usage:\n\tschema_ora.py username passwd tnsentry'

import sys

import cx_Oracle

TABLE_INFO_SQL = """SELECT table_name, column_name, data_type, nullable,
                     decode(default_length, NULL, 0, 1) hasdef,
                     decode(data_type, 'DATE', 11,
                                       'NUMBER', nvl(data_precision,38)||'.'||data_scale,
                                       data_length) data_length
              FROM   user_tab_columns
              ORDER BY 1, 2
"""

PRIMARY_KEYS_INFO_SQL = """SELECT uc.table_name, ucc.column_name
              FROM   user_constraints uc
                    ,user_cons_columns ucc
              WHERE  uc.constraint_name = ucc.constraint_name
              AND    uc.constraint_type = 'P'
              ORDER BY uc.table_name, ucc.position"""


INDEXES_INFO_SQL = """SELECT ui.table_name, ui.index_name, ui.uniqueness
              FROM user_indexes ui
              ORDER BY 1, 2"""


INDEXES_COLUMNS_INFO_SQL = """SELECT table_name, column_name, index_name, column_position, descend
              FROM user_ind_columns
              ORDER BY 1, 2, 3"""


COMPOSITE_INDEXES_COLUMNS_INFO_SQL = """SELECT table_name, column_name, index_name, column_position
              FROM user_ind_columns
              WHERE index_name in (select distinct index_name from USER_IND_COLUMNS where column_position > 1)
              ORDER BY table_name, index_name, column_position
"""


FOREIGN_KEYS_INFO_SQL = """SELECT uc.table_name
                    ,ucc.column_name
                    ,fc.table_name
                    ,fc.constraint_name
              FROM   user_cons_columns ucc
                    ,user_constraints fc
                    ,user_constraints uc
              WHERE  uc.constraint_type = 'R'
              AND    uc.constraint_name = ucc.constraint_name
              AND    fc.constraint_name = uc.r_constraint_name
              ORDER BY 1, 2"""


DEFAULTS_INFO_SQL = """SELECT table_name, column_name, data_default
              FROM   user_tab_columns
              WHERE  default_length IS NOT NULL
              ORDER BY 1, 2"""


TRIGGERS_INFO_SQL = """SELECT trigger_name, trigger_type, triggering_event, table_name, description, trigger_body
              FROM user_triggers"""


_CONN = None

def init_db_conn(username, passwd, tnsentry):
	global _CONN
	dbinfo = 'db: %s@%s' % (username, tnsentry)
	try:
		if not _CONN:
			print dbinfo
			_CONN = cx_Oracle.connect(username, passwd, tnsentry)
	except:
		ex = sys.exc_info()
		s = 'Exception: %s: %s\n%s' % (ex[0], ex[1], dbinfo)
		print s 
		return None
	return _CONN


def db_conn():
	return _CONN


def select_qry(querystr):
	cur = db_conn().cursor()
	cur.execute(querystr)
	results = cur.fetchall()
	cur.close()
	return results


def run_qry(querystr):
	cur = db_conn().cursor()
	cur.execute(querystr)
	cur.close()


def show_qry(title, querystr, fld_join = '\t', row_separator = None):
	rs = select_qry(querystr)
	if rs:
		print '\n\n'
		print '--- %s ---' % title
		for row in rs:
			print fld_join.join([str(s) for s in row])
			if row_separator:
				print row_separator


def init_session():
	run_qry("ALTER SESSION SET nls_numeric_characters = '.,'")


def show_tables():
	show_qry('tables', TABLE_INFO_SQL)


def show_primary_keys():
	show_qry('primary keys', PRIMARY_KEYS_INFO_SQL)


def show_indexes():
	show_qry('indexes', INDEXES_INFO_SQL)
	show_qry('indexes columns', INDEXES_COLUMNS_INFO_SQL)
	show_qry('composite indexes', COMPOSITE_INDEXES_COLUMNS_INFO_SQL)


def show_foreign_keys():
	show_qry('foreign keys', FOREIGN_KEYS_INFO_SQL)


def show_defaults():
	show_qry('defaults', DEFAULTS_INFO_SQL)


def show_procedures():
	print '\n\n --- procedures ---'
	cur = db_conn().cursor()
	cur.execute('SELECT object_name FROM user_procedures ORDER BY 1')
	rows = cur.fetchall()
	for funname in rows:
		print '\n\n -- >>> %s >>> --' % funname
		cur.execute("SELECT text FROM user_source where name = '%s' ORDER BY line" % funname)
		lines = cur.fetchall()
		for line in lines:
			print line[0].rstrip()
		print '\n\n -- <<< %s <<< --' % funname
	cur.close()


def show_triggers():
	show_qry('triggers', TRIGGERS_INFO_SQL, '\n', '\n\n')


def test():
	main('user', 'passwd', 'tns')


def main(username, passwd, tnsentry):
	if not init_db_conn(username, passwd, tnsentry):
		print 'Something is terribly wrong with db connection'
	else:
		init_session()
		show_tables()
		show_primary_keys()
		show_indexes()
		show_foreign_keys()
		show_defaults()
		show_triggers()
		show_procedures()


if '--version' in sys.argv:
	print __version__
elif '--test' in sys.argv:
	test()
elif __name__ == '__main__':
	if len(sys.argv) != 4:
		print USAGE
	else:
		main(*sys.argv[1:])