#!/usr/bin/env python
# -*- coding: utf8 -*-
__version__ = '$Id: schema_ora.py 781 2010-10-26 07:14:26Z 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:])
Diff to Previous Revision
--- revision 6 2009-10-13 23:28:22
+++ revision 7 2010-10-26 07:19:24
@@ -1,6 +1,6 @@
-#!/usr/bin/python
-# -*- coding: cp1250 -*-
-__version__ = '$Id: schema_ora.py 654 2009-10-14 06:11:58Z mn $'
+#!/usr/bin/env python
+# -*- coding: utf8 -*-
+__version__ = '$Id: schema_ora.py 781 2010-10-26 07:14:26Z mn $'
# export Oracle schema to text
# usable to compare databases that should be the same
