#!/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:])