ActiveState Code

#!/usr/bin/python
# -*- coding: cp1250 -*-
__version__ = '$Id: schema_ora.py 654 2009-10-14 06:11:58Z 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

USE_JYTHON = 0

try:
	from com.ziclix.python.sql import zxJDBC
	USE_JYTHON = 1
	USAGE = """usage:
\tschema_ora.py jdbcurl user passwd
example:
\tjython schema_ora.py jdbc:oracle:thin:user/passwd@127.0.0.1:1521:dbname user passwd > db.schema 2> db.err
"""
except:
	USAGE = 'usage:\n\tschema_ora.py user passwd tnsentry'
	USE_JYTHON = 0
	import cx_Oracle

OUT_ENCODINGS = ('cp1250', 'iso8859_2')

TABLE_NAMES_SQL = """SELECT DISTINCT table_name
FROM user_tab_columns
WHERE INSTR(table_name, 'X_') <> 1
ORDER BY 1
"""


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
WHERE INSTR(table_name, 'X_') <> 1
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'
AND INSTR(uc.table_name, 'X_') <> 1
ORDER BY uc.table_name, ucc.position"""


INDEXES_INFO_SQL = """SELECT ui.table_name, ui.index_name, ui.uniqueness
FROM user_indexes ui
WHERE INSTR(ui.table_name, 'X_') <> 1
ORDER BY 1, 2"""


INDEXES_COLUMNS_INFO_SQL = """SELECT table_name, column_name, index_name, column_position, descend
FROM user_ind_columns
WHERE INSTR(table_name, 'X_') <> 1
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)
AND INSTR(table_name, 'X_') <> 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
AND INSTR(uc.table_name, 'X_') <> 1
ORDER BY 1, 2"""


DEFAULTS_INFO_SQL = """SELECT table_name, column_name, data_default
FROM   user_tab_columns
WHERE  default_length IS NOT NULL
AND INSTR(table_name, 'X_') <> 1
ORDER BY table_name, column_name"""


VIEWS_INFO_SQL = """SELECT view_name, text
FROM user_views
ORDER BY view_name"""


TRIGGERS_INFO_SQL = """SELECT trigger_name, trigger_type, triggering_event, table_name, description, trigger_body
FROM user_triggers
WHERE INSTR(table_name, 'X_') <> 1
ORDER BY table_name, trigger_name"""


_CONN = None

def init_db_conn(username, passwd, tnsentry):
	global _CONN
	if not _CONN:
		dbinfo = username
		try:
			if USE_JYTHON:
				print(dbinfo)
				_CONN = zxJDBC.connect(username, passwd, tnsentry, 'oracle.jdbc.driver.OracleDriver')
			else:
				dbinfo = 'db: %s@%s' % (username, tnsentry)
				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 output_line(line):
	try:
		line = line.rstrip()
		for enc in OUT_ENCODINGS:
			try:
				line = line.encode(enc)
				break
			except UnicodeDecodeError:
				pass
		print(line)
	except UnicodeDecodeError:
		print('!!! line cannot be encoded !!!')
		pass


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 fld2str(fld_v):
	if type(fld_v) == type(1.1):
		fld_v = '%s' % fld_v
		if '.' in fld_v:
			fld_v = fld_v.rstrip('0')
			fld_v = fld_v.rstrip('.')
	else:
		fld_v = '%s' % fld_v
		if fld_v.startswith('SYS_C'):
			fld_v = 'SYS_Cxxx'
	return fld_v


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


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


def show_tables():
	show_qry('tables', TABLE_NAMES_SQL)
	show_qry('columns', 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_views():
	show_qry('views', VIEWS_INFO_SQL, '\n', '\n\n')


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:
			output_line(line[0])
		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_views()
		show_triggers()
		show_procedures()
		print('\n\n--- the end ---')


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:])