sqlpython: sqlpyPlus.py comparison

comparison sqlpyPlus.py @ 115:43f5dc75b791

migrating to onecmd from select

author	catherine@Elli.myhome.westell.com
date	Sat, 28 Jun 2008 11:10:09 -0400
parents	c1c05670b4e5
children

comparison

equal deleted inserted replaced

-:c1c05670b4e5
+:43f5dc75b791
 - catherinedevlin.blogspot.com  May 31, 2006
 """
 # note in cmd.cmd about supporting emacs commands?
 descQueries = {
-'TABLE': ("""
+'TABLE': ("""SELECT
 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 '(' ||
 data_type
 FROM all_tab_columns atc
 WHERE atc.table_name = :object_name
 AND      atc.owner = :owner
 ORDER BY atc.column_id;""",),
-'PROCEDURE': ("""
+'PROCEDURE': ("""SELECT
 argument_name,
 data_type,
 in_out,
 default_value
 FROM all_arguments
 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*)\s*$', re.DOTALL | re.MULTILINE)
-# what about quote-enclosed?
 legalOracle = re.compile('[a-zA-Z_$#]')
 def do_select(self, arg, bindVarsIn=None, override_terminator=None):
 """Fetch rows from a table.
 Output may be formatted by choosing an alternative terminator
 ("help terminators" for details)
 """
 bindVarsIn = bindVarsIn or {}
-self.query = 'select ' + arg
+self.query = arg.parent.executable
-terminator = self.commmand_terminator_finder(self.query)
+terminator = arg.parent.terminator or ';'
-if terminator:
+rowlimit = int(arg.parent.terminator_suffix or '0')
-(self.query, terminator, dummy) = terminator
-else:
-terminator = [';']
-try:
-terminator, rowlimit = terminator[0], int(terminator[1])
-except (IndexError, ValueError):
-terminator, rowlimit = terminator[0], 0
 if override_terminator:
 terminator = override_terminator
 try:
 self.varsUsed = findBinds(self.query, self.binds, bindVarsIn)
 self.curs.execute(self.query, self.varsUsed)
 @options([make_option('-f', '--full', action='store_true', help='get dependent objects as well')])
 def do_pull(self, arg, opts):
 """Displays source code."""
-object_type, owner, object_name = self.resolve(arg.strip(self.terminator).upper())
+object_type, owner, object_name = self.resolve(arg.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 opts.insensitive:
 searchfor = "LOWER(text)"
 arg = arg.lower()
 else:
 searchfor = "text"
-self.do_select("* from all_source where %s like '%%%s%%'" % (searchfor, arg))
+self.onecmd("select * from all_source where %s like '%%%s%%';" % (searchfor, arg))
 @options([make_option('-a','--all',action='store_true',
 help='Describe all objects (not just my own)')])
 def do_describe(self, arg, opts):
 "emulates SQL*Plus's DESCRIBE"
 which_view = (', owner', 'all')
 else:
 which_view = ('', 'user')
 if not arg:
-self.do_select("""object_name, object_type%s FROM %s_objects WHERE object_type IN ('TABLE','VIEW','INDEX') ORDER BY object_name""" % which_view)
+self.onecmd("""SELECT object_name, object_type%s FROM %s_objects WHERE object_type IN ('TABLE','VIEW','INDEX') ORDER BY object_name;""" % which_view)
 return
-object_type, owner, object_name = self.resolve(arg.strip(self.terminator).upper())
+object_type, owner, object_name = self.resolve(arg.upper())
 if not object_type:
-self.do_select("""object_name, object_type%s FROM %s_objects
+self.onecmd("""SELECT object_name, object_type%s FROM %s_objects
 WHERE object_type IN ('TABLE','VIEW','INDEX')
 AND   object_name LIKE '%%%s%%'
-ORDER BY object_name""" %
+ORDER BY object_name;""" %
 (which_view[0], which_view[1], 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:
-self.do_select(q,bindVarsIn={'object_name':object_name, 'owner':owner})
+self.onecmd(q,bindVarsIn={'object_name':object_name, 'owner':owner})
 elif object_type == 'PACKAGE':
 self.curs.execute(descQueries['PackageObjects'][0], {'package_name':object_name, 'owner':owner})
 packageContents = self.curs.fetchall()
 for (packageObj_name,) in packageContents:
 self.stdout.write(packageObj_name + '\n')
-self.do_select(descQueries['PackageObjArgs'][0],bindVarsIn={'package_name':object_name, 'owner':owner, 'object_name':packageObj_name})
+self.onecmd(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())
+object_type, owner, object_name = self.resolve(arg.upper())
 if object_type == 'PACKAGE BODY':
 q = "and (type != 'PACKAGE BODY' or name != :object_name)'"
 object_type = 'PACKAGE'
 else:
 q = ""
-q = """         name,
+q = """SELECT
+name,
 type
 from user_dependencies
 where
 referenced_name like :object_name
 and	referenced_type like :object_type
 and	referenced_owner like :owner
-%s""" % (q)
+%s;""" % (q)
-self.do_select(q, {'object_name':object_name, 'object_type':object_type, 'owner':owner})
+self.onecmd(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())
+object_type, owner, object_name = self.resolve(arg.upper())
 if object_type:
 self.curs.execute(queries['tabComments'],{'table_name':object_name, 'owner':owner})
 self.stdout.write("%s %s.%s: %s\n" % (object_type, owner, object_name, self.curs.fetchone()[0]))
-self.do_select(queries['colComments'],bindVarsIn={'owner':owner, 'object_name': object_name})
+self.onecmd(queries['colComments'],bindVarsIn={'owner':owner, 'object_name': object_name})
 def resolve(self, identifier):
 """Checks (my objects).name, (my synonyms).name, (public synonyms).name
 to resolve a database object's name. """
 parts = identifier.split('.')
 print self.do_compare.__doc__
 return
 for n in range(len(args2)):
 query = args2[n]
 fnames.append('compare%s.txt' % n)
-if query.rstrip()[-1] != self.terminator:
+if query.rstrip()[-1] != self.terminator: #TODO: fix with new terminators
 query = '%s%s' % (query, self.terminator)
 self.onecmd_plus_hooks('%s > %s' % (query, fnames[n]))
 diffMergeSearcher.invoke(fnames[0], fnames[1])
 bufferPosPattern = re.compile('\d+')
 def do__dir_tables(self, arg, opts):
 if opts.all:
 which_view = (', owner', 'all')
 else:
 which_view = ('', 'user')
-self.do_select("""table_name, 'TABLE' as type%s FROM %s_tables WHERE table_name LIKE '%%%s%%'""" %
+self.onecmd("""SELECT table_name, 'TABLE' as type%s FROM %s_tables WHERE table_name LIKE '%%%s%%';""" %
 (which_view[0], which_view[1], arg.upper()))
 @options([make_option('-a','--all',action='store_true',
 help='Describe all objects (not just my own)')])
 def do__dir_views(self, arg, opts):
 if opts.all:
 which_view = (', owner', 'all')
 else:
 which_view = ('', 'user')
-self.do_select("""view_name, 'VIEW' as type%s FROM %s_views WHERE view_name LIKE '%%%s%%'""" %
+self.onecmd("""SELECT view_name, 'VIEW' as type%s FROM %s_views WHERE view_name LIKE '%%%s%%';""" %
 (which_view[0], which_view[1], arg.upper()))
 @options([make_option('-a','--all',action='store_true',
 help='Describe all objects (not just my own)')])
 def do__dir_indexes(self, arg, opts):
 if opts.all:
 which_view = (', owner', 'all')
 else:
 which_view = ('', 'user')
-self.do_select("""index_name, index_type%s FROM %s_indexes WHERE index_name LIKE '%%%s%%' OR table_name LIKE '%%%s%%'""" %
+self.onecmd("""SELECT index_name, index_type%s FROM %s_indexes WHERE index_name LIKE '%%%s%%' OR table_name LIKE '%%%s%%';""" %
 (which_view[0], which_view[1], arg.upper(), arg.upper()))
 def do__dir_tablespaces(self, arg):
-self.do_select("""tablespace_name, file_name from dba_data_files""")
+self.onecmd("""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""")
+self.onecmd("""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:
 nrows = int(a[1:])
 args.remove(a)
 except:
 pass
 arg = ' '.join(args)
-self.do_select('* from %s;%d' % (arg, nrows))
+self.onecmd('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] == ':':
 self.stdout.write('\n'.join(result) + '\n')
 def do_cat(self, arg):
 targets = arg.split()
 for target in targets:
-self.do_select('* from %s' % target)
+self.onecmd('select * from %s;' % target)
 @options([make_option('-i', '--ignore-case', dest='ignorecase', action='store_true', help='Case-insensitive search')])
 def do_grep(self, arg, opts):
 """grep PATTERN TABLE - search for term in any of TABLE's fields"""
 self.curs.execute('select * from %s where 1=0' % target)
 if opts.ignorecase:
 sql = ' or '.join("LOWER(%s) LIKE '%%%s%%'" % (d[0], pattern.lower()) for d in self.curs.description)
 else:
 sql = ' or '.join("%s LIKE '%%%s%%'" % (d[0], pattern) for d in self.curs.description)
-sql = '* FROM %s WHERE %s' % (target, sql)
+sql = 'SELECT * FROM %s WHERE %s;' % (target, sql)
-self.do_select(sql)
+self.onecmd(sql)
 except Exception, e:
 print e
 import traceback
 traceback.print_exc(file=sys.stdout)
 def do_refs(self, arg):
-object_type, owner, object_name = self.resolve(arg.strip(self.terminator).upper())
+object_type, owner, object_name = self.resolve(arg.upper())
 if object_type == 'TABLE':
-self.do_select(queries['refs'],bindVarsIn={'object_name':object_name, 'owner':owner})
+self.onecmd(queries['refs'],bindVarsIn={'object_name':object_name, 'owner':owner})
 def _test():
 import doctest
 doctest.testmod()

Mercurial > sqlpython

comparison sqlpyPlus.py @ 115:43f5dc75b791