sqlpython: sqlpyPlus.py comparison

comparison sqlpyPlus.py @ 80:83de0cb04f12

prevent crash on lone backslash

author	catherine@localhost
date	Wed, 14 May 2008 17:23:52 -0400
parents	01d578f4e6e7
children	32c868fca272

comparison

equal deleted inserted replaced

-:01d578f4e6e7
+:83de0cb04f12
 - SQL*Plus-style describe, spool
 - write sends query result directly to file
 - comments shows table and column comments
 - compare ... to ... graphically compares results of two queries
 - commands are case-insensitive
 Use 'help' within sqlpython for details.
 Compatible with sqlpython v1.3
 Set bind variables the hard (SQL*Plus) way
 """
 # note in cmd.cmd about supporting emacs commands?
 descQueries = {
 'TABLE': ("""
 atc.column_name,
 CASE atc.nullable WHEN 'Y' THEN 'NULL' ELSE 'NOT NULL' END "Null?",
 atc.data_type ||
 CASE atc.data_type WHEN 'DATE' THEN ''
 ELSE '(' ||
 CASE atc.data_type WHEN 'NUMBER' THEN TO_CHAR(atc.data_precision) ||
 CASE atc.data_scale WHEN 0 THEN ''
 ELSE ',' || TO_CHAR(atc.data_scale) END
 ELSE TO_CHAR(atc.data_length) END
 END ||
 CASE atc.data_type WHEN 'DATE' THEN '' ELSE ')' END
 data_type
 FROM all_tab_columns atc
 WHERE atc.table_name = :object_name
 AND      atc.owner = :owner
 ORDER BY atc.column_id;""",),
 'PROCEDURE': ("""
 argument_name,
 data_type,
 in_out,
 default_value
 FROM all_arguments
 WHERE object_name = :object_name
 AND      owner = :owner
 AND      package_name IS NULL
 AND      argument_name IS NOT NULL
 SELECT DISTINCT object_name
 FROM all_arguments
 WHERE package_name = :package_name
 AND      owner = :owner""",),
 'PackageObjArgs':("""
 object_name,
 argument_name,
 data_type,
 in_out,
 default_value
 FROM all_arguments
 WHERE package_name = :package_name
 AND      object_name = :object_name
 AND      owner = :owner
 AND      argument_name IS NOT NULL
 ORDER BY sequence""",),
 'TRIGGER':("""
 description
 FROM   all_triggers
 WHERE  owner = :owner
 AND    trigger_name = :object_name
 """,
 """
 table_owner,
 base_object_type,
 table_name,
 column_name,
 when_clause,
 status,
 action_type,
 crossedition
 FROM   all_triggers
 WHERE  owner = :owner
 AND    trigger_name = :object_name
 \\t
 """,
 FROM   all_indexes
 WHERE  owner = :owner
 AND    index_name = :object_name
 \\t
 """,)
 }
 descQueries['VIEW'] = descQueries['TABLE']
 descQueries['FUNCTION'] = descQueries['PROCEDURE']
 queries = {
 'resolve': """
 SELECT object_type, object_name, owner FROM (
 SELECT object_type, object_name, user owner, 1 priority
 FROM   user_objects
 WHERE object_name = :objName
 UNION ALL
 SELECT ao.object_type, ao.object_name, ao.owner, 2 priority
 FROM    all_objects ao
 JOIN      user_synonyms us ON (us.table_owner = ao.owner AND us.table_name = ao.object_name)
 WHERE us.synonym_name = :objName
 AND   ao.object_type != 'SYNONYM'
 UNION ALL
 SELECT ao.object_type, ao.object_name, ao.owner, 3 priority
 FROM    all_objects ao
 JOIN      all_synonyms asyn ON (asyn.table_owner = ao.owner AND asyn.table_name = ao.object_name)
 WHERE asyn.synonym_name = :objName
 AND   ao.object_type != 'SYNONYM'
 AND      asyn.owner = 'PUBLIC'
 UNION ALL
 SELECT 'DIRECTORY' object_type, dir.directory_name, dir.owner, 6 priority
 FROM   all_directories dir
 WHERE  dir.directory_name = :objName
 UNION ALL
 SELECT 'DATABASE LINK' object_type, db_link, owner, 7 priority
 FROM   all_db_links dbl
 WHERE  dbl.db_link = :objName
 ) ORDER BY priority ASC""",
 'tabComments': """
 SELECT comments
 FROM    all_tab_comments
 WHERE owner = :owner
 AND      table_name = :table_name""",
 'colComments': """
 atc.column_name,
 acc.comments
 FROM all_tab_columns atc
 JOIN all_col_comments acc ON (atc.owner = acc.owner and atc.table_name = acc.table_name and atc.column_name = acc.column_name)
 WHERE atc.table_name = :object_name
 AND      atc.owner = :owner
 ORDER BY atc.column_id;""",
 #thanks to Senora.pm for "refs"
 'refs': """
 NULL               referenced_by,
 c2.table_name      references,
 c1.constraint_name constraint
 FROM
 user_constraints c1,
 user_constraints c2
 WHERE
 c1.table_name = :object_name
 and c1.constraint_type ='R'
 and c1.r_constraint_name = c2.constraint_name
 and c1.r_owner = c2.owner
 and c1.owner = :owner
 UNION
 SELECT c1.table_name      referenced_by,
 NULL               references,
 c1.constraint_name constraint
 FROM
 user_constraints c1,
 user_constraints c2
 WHERE
 c2.table_name = :object_name
 and c1.constraint_type ='R'
 and c1.r_constraint_name = c2.constraint_name
 and c1.r_owner = c2.owner
 and c1.owner = :owner
 """
 }
 import sys, os, re, sqlpython, cx_Oracle, pyparsing
 from cmd2 import flagReader, Cmd
 if float(sys.version[:3]) < 2.3:
 def enumerate(lst):
 return zip(range(len(lst)), lst)
 class SoftwareSearcher(object):
 def __init__(self, softwareList, purpose):
 self.softwareList = softwareList
 self.purpose = purpose
 self.software = None
 return (software, '%s %s')
 for (n, (software, invokeString)) in enumerate(self.softwareList):
 if os.path.exists(software):
 if n > (len(self.softwareList) * 0.7):
 print """
 Using %s.  Note that there are better options available for %s,
 but %s couldn't find a better one in your PATH.
 Feel free to open up %s
 and customize it to find your favorite %s program.
 """ % (software, self.purpose, __file__, __file__, self.purpose)
 return (software, invokeString)
 stem = os.path.split(software)[1]
 for p in os.environ['PATH'].split(os.pathsep):
 if os.path.exists(os.sep.join([p, stem])):
 return (stem, invokeString)
 raise (OSError, """Could not find any %s programs.  You will need to install one,
 or customize %s to make it aware of yours.
 Looked for these programs:
 %s""" % (self.purpose, __file__, "\n".join([s[0] for s in self.softwareList])))
 #v2.4: %s""" % (self.purpose, __file__, "\n".join(s[0] for s in self.softwareList)))
 softwareLists = {
 'diff/merge': [
 ('/usr/bin/meld',"%s %s %s"),
 ('/usr/bin/kdiff3',"%s %s %s"),
 (r'C:\Program Files\Araxis\Araxis Merge v6.5\Merge.exe','"%s" %s %s'),
 (r'C:\Program Files\TortoiseSVN\bin\TortoiseMerge.exe', '"%s" /base:"%s" /mine:"%s"'),
 ('FileMerge','%s %s %s'),
 ('kompare','%s %s %s'),
 ('WinMerge','%s %s %s'),
 ('xxdiff','%s %s %s'),
 ('fldiff','%s %s %s'),
 ('gtkdiff','%s %s %s'),
 ('tkdiff','%s %s %s'),
 ('gvimdiff','%s %s %s'),
 ('diff',"%s %s %s"),
 (r'c:\windows\system32\comp.exe',"%s %s %s")],
 'text editor': [
 ('gedit', '%s %s'),
 ('textpad', '%s %s'),
 ('notepad.exe', '%s %s'),
 ('pico', '%s %s'),
 ('emacs', '%s %s'),
 ('vim', '%s %s'),
 ('vi', '%s %s'),
 ('ed', '%s %s'),
 ('edlin', '%s %s')
 ]
 }
 diffMergeSearcher = SoftwareSearcher(softwareLists['diff/merge'],'diff/merge')
 editSearcher = SoftwareSearcher(softwareLists['text editor'], 'text editor')
 editor = os.environ.get('EDITOR')
 result[varname] = existingBinds[varname]
 except KeyError:
 if not givenBindVars.has_key(varname):
 print 'Bind variable %s not defined.' % (varname)
 return result
 class sqlpyPlus(sqlpython.sqlpython):
 defaultExtension = 'sql'
 shortcuts = {'?': 'help', '@': 'getrun', '!': 'shell', ':': 'setbind', '\\': 'psql'}
 multilineCommands = '''select insert update delete tselect
 create drop alter'''.split()
 defaultFileName = 'afiedt.buf'
 def __init__(self):
 sqlpython.sqlpython.__init__(self)
 self.binds = CaselessDict()
 self.sqlBuffer = []
 def parseline(self, line):
 """Parse the line into a command name and a string containing
 the arguments.  Returns a tuple containing (command, args, line).
 'command' and 'args' may be None if the line couldn't be parsed.
 Overrides cmd.cmd.parseline to accept variety of shortcuts.."""
 cmd, arg, line = sqlpython.sqlpython.parseline(self, line)
 if cmd in ('select', 'sleect', 'insert', 'update', 'delete', 'describe',
 'desc', 'comments', 'pull', 'refs', 'desc', 'triggers', 'find') \
 and not hasattr(self, 'curs'):
 print 'Not connected.'
 return '', '', ''
 return cmd, arg, line
 def onecmd_plus_hooks(self, line):
 line = self.precmd(line)
 stop = self.onecmd(line)
 stop = self.postcmd(stop, line)
 for (itm, useq) in zip(row, usequotes))
 result = ['INSERT INTO %s (%s) VALUES (%s);' %
 (self.tblname, ','.join(self.colnames), formatRow(row))
 for row in self.rows]
 return '\n'.join(result)
 def output_row_as_xml(self, row):
 result = ['  <%s>\n    %s\n  </%s>' %
 (colname.lower(), str('' if (itm is None) else itm), colname.lower())
 for (itm, colname) in zip(row, self.colnames)]
 return '\n'.join(result)
 result = ['  <tr>\n    %s\n  </tr>' % result]
 for row in self.rows:
 result.append('  <tr>\n    %s\n  </tr>' %
 (''.join('<td>%s</td>' %
 str('' if (itm is None) else itm)
 for itm in row)))
 result = '''<table id="%s">
 %s
 </table>''' % (self.tblname, '\n'.join(result))
 return result
 newdesc[0][0] = 'COLUMN NAME'
 result = '\n' + sqlpython.pmatrix(transpr,newdesc)
 else:
 result = sqlpython.pmatrix(self.rows, self.curs.description, self.maxfetch)
 return result
 statementEndPattern = re.compile(r'(.*)(;|\\[gGhtxicCsS])\s*(\d*)$', re.DOTALL | re.MULTILINE)
 # what about quote-enclosed?
 def findTerminator(self, statement):
 m = self.statementEndPattern.search(statement)
 if m:
 return m.groups()
 else:
 return statement, None, None
 legalOracle = re.compile('[a-zA-Z_$#]')
 def do_select(self, arg, bindVarsIn=None, override_terminator=None):
 """Fetch rows from a table.
 Limit the number of rows retrieved by appending
 an integer after the terminator
 (example: SELECT * FROM mytable;10 )
 Output may be formatted by choosing an alternative terminator
 ("help terminators" for details)
 """
 bindVarsIn = bindVarsIn or {}
 self.query = 'select ' + arg
 self.sqlBuffer.append(self.query)
 pullflags = flagReader.FlagSet([flagReader.Flag('full')])
 def do_pull(self, arg):
 """Displays source code.
 --full, -f: get dependent objects as well"""
 options, arg = self.pullflags.parse(arg)
 object_type, owner, object_name = self.resolve(arg.strip(self.terminator).upper())
 if not object_type:
 return
 self.stdout.write("%s %s.%s\n" % (object_type, owner, object_name))
 self.stdout.write(str(self.curs.callfunc('DBMS_METADATA.GET_DDL', cx_Oracle.CLOB,
 [object_type, object_name, owner])))
 if options.has_key('full'):
 for dependent_type in ('OBJECT_GRANT', 'CONSTRAINT', 'TRIGGER'):
 try:
 self.stdout.write(str(self.curs.callfunc('DBMS_METADATA.GET_DEPENDENT_DDL', cx_Oracle.CLOB,
 [dependent_type, object_name, owner])))
 except cx_Oracle.DatabaseError:
 pass
 findflags = flagReader.FlagSet([flagReader.Flag('insensitive')])
 def do_find(self, arg):
 """Finds argument in source code.
 --insensitive, -i: case-insensitive search"""
 options, arg = self.findflags.parse(arg)
 if options.has_key('insensitive'):
 searchfor = "LOWER(text)"
 arg = arg.lower()
 else:
 searchfor = "text"
 self.do_select("* from all_source where %s like '%%%s%%'" % (searchfor, arg))
 def do_describe(self, arg):
 "emulates SQL*Plus's DESCRIBE"
 if not arg:
-self.do_select("""object_name, object_type FROM all_objects WHERE object_type IN ('TABLE','VIEW','INDEX') ORDER BY object_name""")
+self.do_select("""object_name, object_type, owner FROM all_objects WHERE object_type IN ('TABLE','VIEW','INDEX') ORDER BY object_name""")
 return
 object_type, owner, object_name = self.resolve(arg.strip(self.terminator).upper())
 if not object_type:
-self.do_select("""object_name, object_type FROM all_objects
+self.do_select("""object_name, object_type, owner FROM all_objects
 WHERE object_type IN ('TABLE','VIEW','INDEX')
 AND   object_name LIKE '%%%s%%'
 ORDER BY object_name""" % arg.upper() )
 return
 self.stdout.write("%s %s.%s\n" % (object_type, owner, object_name))
 descQ = descQueries.get(object_type)
 if descQ:
 for q in descQ:
 packageContents = self.curs.fetchall()
 for (packageObj_name,) in packageContents:
 print packageObj_name
 self.do_select(descQueries['PackageObjArgs'][0],bindVarsIn={'package_name':object_name, 'owner':owner, 'object_name':packageObj_name})
 do_desc = do_describe
+def do_deps(self, arg):
+object_type, owner, object_name = self.resolve(arg.strip(self.terminator).upper())
+if object_type == 'PACKAGE BODY':
+q = "and (type != 'PACKAGE BODY' or name != :object_name)'"
+object_type = 'PACKAGE'
+else:
+q = ""
+q = """         name,
+type
+from user_dependencies
+where
+referenced_name like :object_name
+and	referenced_type like :object_type
+and	referenced_owner like :owner
+%s""" % (q)
+self.do_select(q, {'object_name':object_name, 'object_type':object_type, 'owner':owner})
 def do_comments(self, arg):
 'Prints comments on a table and its columns.'
 object_type, owner, object_name = self.resolve(arg.strip(self.terminator).upper())
 if object_type:
 self.curs.execute(queries['tabComments'],{'table_name':object_name, 'owner':owner})
 object_type, owner, object_name = '', '', ''
 return object_type, owner, object_name
 def do_resolve(self, arg):
 self.stdout.write(self.resolve(arg)+'\n')
 def spoolstop(self):
 if self.spoolFile:
 sys.stdout = self.stdoutBeforeSpool
 print 'Finished spooling to ', self.spoolFile.name
 self.spoolFile.close()
 self.spoolFile = None
 def do_spool(self, arg):
 """spool [filename] - begins redirecting output to FILENAME."""
 self.spoolstop()
 arg = arg.strip()
 if not arg:
 f = open(fname, 'w')
 sys.stdout = f
 self.onecmd_plus_hooks(arg)
 f.close()
 sys.stdout = originalOut
 def do_write(self, args):
 'write [filename.extension] query - writes result to a file'
 words = args.split(None, 1)
 if len(words) > 1 and '.' in words[0]:
 fname, command = words
 else:
 fname, command = 'output.txt', args
 self.write(command, fname)
 print 'Results written to %s' % os.path.join(os.getcwd(), fname)
 def do_compare(self, args):
 """COMPARE query1 TO query2 - uses external tool to display differences.
 Sorting is recommended to avoid false hits.
 Will attempt to use a graphical diff/merge tool like kdiff3, meld, or Araxis Merge,
 if they are installed."""
 fnames = []
 args2 = args.split(' to ')
 def do_getrun(self, fname):
 'Brings SQL commands from a file to the in-memory SQL buffer, and executes them.'
 Cmd.do_load(self, fname)
 def do_psql(self, arg):
 '''Shortcut commands emulating psql's backslash commands.
 \c connect
 \d desc
 \e edit
 \g run
 \h help
 commands = {}
 for c in self.do_psql.__doc__.splitlines()[2:]:
 (abbrev, command) = c.split(None, 1)
 commands[abbrev[1:]] = command
 words = arg.split(None,1)
-abbrev = words[0]
+try:
+abbrev = words[0]
+except IndexError:
+return
 try:
 args = words[1]
 except IndexError:
 args = ''
 try:
 except KeyError:
 print 'psql command \%s not yet supported.' % abbrev
 def do__dir_tables(self, arg):
 self.do_select("""table_name, 'TABLE' as type, owner FROM all_tables WHERE table_name LIKE '%%%s%%'""" % arg.upper())
 def do__dir_views(self, arg):
 self.do_select("""view_name, 'VIEW' as type, owner FROM all_views WHERE view_name LIKE '%%%s%%'""" % arg.upper())
 def do__dir_indexes(self, arg):
 self.do_select("""index_name, index_type, owner FROM all_indexes WHERE index_name LIKE '%%%s%%' OR table_name LIKE '%%%s%%'""" % (arg.upper(), arg.upper()))
 def do__dir_tablespaces(self, arg):
 self.do_select("""tablespace_name, file_name from dba_data_files""")
 def do__dir_schemas(self, arg):
 self.do_select("""owner, count(*) AS objects FROM all_objects GROUP BY owner ORDER BY owner""")
 def do_head(self, arg):
 nrows = 10
 args = arg.split()
 if len(args) > 1:
 for a in args:
 args.remove(a)
 except:
 pass
 arg = ' '.join(args)
 self.do_select('* from %s;%d' % (arg, nrows))
 def do_print(self, arg):
 'print VARNAME: Show current value of bind variable VARNAME.'
 if arg:
 if arg[0] == ':':
 arg = arg[1:]
 except KeyError:
 self.stdout.write('No bind variable %s\n' % arg)
 else:
 for (var, val) in self.binds.items():
 print ':%s = %s' % (var, val)
 def do_setbind(self, arg):
 args = arg.split(None, 2)
 if len(args) < 2:
 self.do_print(arg)
 elif len(args) > 2 and args[1] in ('=',':='):
 var, val = args[0], args[2]
 if val[0] == val[-1] == "'" and len(val) > 1:
-val = val[1:-1]
+self.binds[var] = val[1:-1]
+return
 try:
-val = int(val)
+self.binds[var] = int(val)
+return
 except ValueError:
 try:
-val = float(val)
+self.binds[var] = float(val)
+return
 except ValueError:
-val = self.curs.callfunc(val, [])
+try:
-# submit to sql
+self.binds[var] = self.curs.callfunc(val, [])
+return
-self.binds[var] = val # but what if val is a function call?
+except:
-else:
+pass
-print 'Could not parse ', args
+# TODO: fix
+print 'Could not parse ', args
 def do_exec(self, arg):
 if arg[0] == ':':
 self.do_setbind(arg[1:])
 else:
 try:
 self.curs.execute('\n'.join(lines))
 except Exception, e:
 print e
 return
 lines.append(line)
 def do_begin(self, arg):
 self.anon_plsql('begin ' + arg)
 def do_declare(self, arg):
 self.anon_plsql('declare ' + arg)
 def do_create(self, arg):
 self.anon_plsql('create ' + arg)
 lsflags = flagReader.FlagSet([flagReader.Flag('long')])
 def do_ls(self, arg):
 options, arg = self.lsflags.parse(arg)
 where = ''
 if arg:
 where = """\nWHERE object_type || '/' || object_name
 LIKE '%%%s%%'""" % (arg.upper().replace('*','%'))
 else:
 where = ''
 result = []
 statement = '''SELECT object_type, object_name,
 status, last_ddl_time
 FROM   user_objects %s
 ORDER BY object_type, object_name''' % (where)
 self.curs.execute(statement)
 for (object_type, object_name, status, last_ddl_time) in self.curs.fetchall():
 if options.has_key('long'):
 result.append('%s\t%s\t%s/%s' % (status, last_ddl_time, object_type, object_name))
 else:
 result.append('%s/%s' % (object_type, object_name))
 self.stdout.write('\n'.join(result) + '\n')
 if options.has_key('insensitive'):
 searchfor = "LOWER(text)"
 arg = arg.lower()
 def do_cat(self, arg):
 targets = arg.split()
 for target in targets:
 self.do_select('* from %s' % target)
 def do_grep(self, arg):
 """grep PATTERN TABLE - search for term in any of TABLE's fields"""
 targets = arg.split()
 pattern = targets.pop(0)
 for target in targets:
 self.do_select(queries['refs'],bindVarsIn={'object_name':object_name, 'owner':owner})
 def _test():
 import doctest
 doctest.testmod()
 if __name__ == "__main__":
 "Silent return implies that all unit tests succeeded.  Use -v to see details."
 _test()

Mercurial > sqlpython

comparison sqlpyPlus.py @ 80:83de0cb04f12