<s> import cmd <EOL> import sys <EOL> from conary import dbstore <EOL> import sqlerrors <EOL> import os <EOL> import itertools <EOL> import time <EOL> try : <EOL> import readline <EOL> except ImportError : <EOL> hasReadline = False <EOL> else : <EOL> hasReadline = True <EOL> class DbShell ( cmd . Cmd ) : <EOL> _history_path = os . path . expanduser ( '<STR_LIT>' ) <EOL> yes_args = ( '<STR_LIT>' , '<STR_LIT:yes>' ) <EOL> no_args = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> sqlstarters = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> sqlkeywords = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:left>' , '<STR_LIT>' , '<STR_LIT:index>' , <EOL> '<STR_LIT>' , '<STR_LIT:null>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> prompt = '<STR_LIT>' <EOL> multiline_prompt = '<STR_LIT>' <EOL> doc_header = "<STR_LIT>" <EOL> intro = """<STR_LIT>""" <EOL> identchars = cmd . IDENTCHARS + '<STR_LIT:.>' <EOL> default_width = <NUM_LIT:10> <EOL> def __init__ ( self , db = None , driver = None , path = None ) : <EOL> cmd . Cmd . __init__ ( self ) <EOL> self . show_headers = False <EOL> self . format = self . format_list <EOL> self . manual_widths = { } <EOL> self . use_pager = False <EOL> self . auto_width = False <EOL> self . show_stats = True <EOL> if driver and path : <EOL> self . db = dbstore . connect ( path , driver = driver ) <EOL> elif db : <EOL> self . db = db <EOL> else : <EOL> raise RuntimeError , '<STR_LIT>' <EOL> self . cu = self . db . cursor ( ) <EOL> def multiline ( self , firstline = '<STR_LIT>' ) : <EOL> full_input = [ ] <EOL> old_hist = [ ] <EOL> if firstline : <EOL> full_input . append ( firstline ) <EOL> while True : <EOL> if hasReadline : <EOL> old_hist . append ( readline . get_current_history_length ( ) ) <EOL> if self . use_rawinput : <EOL> try : <EOL> line = raw_input ( self . multiline_prompt ) <EOL> except EOFError : <EOL> line = '<STR_LIT>' <EOL> else : <EOL> self . stdout . write ( self . multiline_prompt ) <EOL> self . stdout . flush ( ) <EOL> line = self . stdin . readline ( ) <EOL> if not len ( line ) : <EOL> line = '<STR_LIT>' <EOL> else : <EOL> line = line [ : - <NUM_LIT:1> ] <EOL> if line == '<STR_LIT>' : <EOL> print <EOL> break <EOL> full_input . append ( line ) <EOL> if '<STR_LIT:;>' in line : <EOL> break <EOL> if hasReadline : <EOL> old_hist . append ( readline . get_current_history_length ( ) ) <EOL> cmd = '<STR_LIT:U+0020>' . join ( full_input ) <EOL> if hasReadline : <EOL> old_hist = sorted ( set ( old_hist ) ) <EOL> for pos in reversed ( old_hist ) : <EOL> readline . remove_history_item ( pos - <NUM_LIT:1> ) <EOL> readline . add_history ( cmd ) <EOL> return cmd <EOL> def parseline ( self , line ) : <EOL> line = line . strip ( ) <EOL> if line . startswith ( '<STR_LIT:.>' ) : <EOL> line = '<STR_LIT:_>' + line [ <NUM_LIT:1> : ] <EOL> return cmd . Cmd . parseline ( self , line ) <EOL> def calculate_widths ( self , cu ) : <EOL> if self . show_headers : <EOL> widths = [ len ( s ) + <NUM_LIT:4> for s in cu . fields ( ) ] <EOL> else : <EOL> widths = [ <NUM_LIT:0> ] * len ( cu . fields ( ) ) <EOL> rows = [ ] <EOL> for row in cu : <EOL> for pos , col in enumerate ( row ) : <EOL> widths [ pos ] = max ( widths [ pos ] , len ( str ( col ) ) ) <EOL> rows . append ( row ) <EOL> return widths , rows <EOL> def format_column_rows ( self , headers , rows , widths = None ) : <EOL> if not widths : <EOL> if self . show_headers : <EOL> widths = [ len ( s ) + <NUM_LIT:4> for s in headers ] <EOL> else : <EOL> widths = [ self . default_width ] * len ( headers ) <EOL> for col , width in enumerate ( widths ) : <EOL> if col in self . manual_widths : <EOL> widths [ col ] = self . manual_widths [ col ] <EOL> total = sum ( widths ) + ( len ( widths ) - <NUM_LIT:1> ) <EOL> bar = '<STR_LIT:+>' . join ( '<STR_LIT:->' * ( x + <NUM_LIT:2> ) for x in widths ) <EOL> bar = '<STR_LIT:+>' + bar + '<STR_LIT:+>' <EOL> format = '<STR_LIT:|>' . join ( ( '<STR_LIT>' % ( x , x ) ) for x in widths ) <EOL> format = '<STR_LIT:|>' + format + '<STR_LIT:|>' <EOL> yield <NUM_LIT:0> , bar <EOL> if self . show_headers : <EOL> yield <NUM_LIT:0> , format % tuple ( '<STR_LIT>' % x for x in headers ) <EOL> yield <NUM_LIT:0> , bar <EOL> for row in rows : <EOL> yield <NUM_LIT:1> , format % tuple ( row ) <EOL> yield <NUM_LIT:0> , bar <EOL> def format_column ( self , cu ) : <EOL> if self . auto_width : <EOL> widths , rows = self . calculate_widths ( cu ) <EOL> return self . format_column_rows ( cu . fields ( ) , rows , widths ) <EOL> return self . format_column_rows ( cu . fields ( ) , cu ) <EOL> def format_val ( self , col ) : <EOL> if isinstance ( col , float ) : <EOL> return '<STR_LIT>' % col <EOL> return str ( col ) <EOL> def format_list ( self , cu ) : <EOL> fields = None <EOL> for row in cu : <EOL> if self . show_headers and not fields : <EOL> fields = cu . fields ( ) <EOL> yield <NUM_LIT:0> , '<STR_LIT:|>' . join ( fields ) <EOL> yield <NUM_LIT:1> , '<STR_LIT:|>' . join ( self . format_val ( x ) for x in row ) <EOL> def display ( self , cu ) : <EOL> lines = self . format ( cu ) <EOL> if self . use_pager : <EOL> import pydoc <EOL> rows = [ x for x in lines ] <EOL> text = '<STR_LIT:\n>' . join ( x [ <NUM_LIT:1> ] for x in rows ) <EOL> rows = sum ( x [ <NUM_LIT:0> ] for x in rows ) <EOL> end = time . time ( ) <EOL> pydoc . pager ( text ) <EOL> else : <EOL> rows = <NUM_LIT:0> <EOL> for isrow , line in lines : <EOL> if isrow : <EOL> rows += <NUM_LIT:1> <EOL> print line <EOL> end = time . time ( ) <EOL> return rows , end <EOL> def default ( self , cmd ) : <EOL> cmd = cmd . strip ( ) <EOL> if cmd == '<STR_LIT>' : <EOL> print <EOL> return True <EOL> if not cmd . endswith ( '<STR_LIT:;>' ) : <EOL> cmd = self . multiline ( cmd ) . strip ( ) <EOL> if '<STR_LIT:;>' in cmd : <EOL> complete , partial = cmd . split ( '<STR_LIT:;>' , <NUM_LIT:1> ) <EOL> if partial : <EOL> self . default ( complete + '<STR_LIT:;>' ) <EOL> return self . default ( partial ) <EOL> if not cmd : <EOL> return False <EOL> start = time . time ( ) <EOL> try : <EOL> if r'<STR_LIT>' in cmd : <EOL> cmd = cmd % self . db . keywords <EOL> self . cu . execute ( cmd ) <EOL> except sqlerrors . DatabaseError , e : <EOL> if len ( e . args ) > <NUM_LIT:1> : <EOL> print '<STR_LIT>' , str ( e . args [ <NUM_LIT:0> ] ) <EOL> else : <EOL> print '<STR_LIT>' , str ( e ) <EOL> return False <EOL> except Exception , e : <EOL> print '<STR_LIT>' , str ( e ) <EOL> return False <EOL> if self . cu . fields ( ) is None or not len ( self . cu . fields ( ) ) : <EOL> print "<STR_LIT>" <EOL> self . db . loadSchema ( ) <EOL> else : <EOL> rows , end = self . display ( self . cu ) <EOL> if self . show_stats and rows != - <NUM_LIT:1> : <EOL> print '<STR_LIT>' % ( rows , end - start ) <EOL> return False <EOL> def cmdloop ( self ) : <EOL> self . read_history ( ) <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> rc = cmd . Cmd . cmdloop ( self ) <EOL> except KeyboardInterrupt : <EOL> self . intro = None <EOL> print <EOL> continue <EOL> break <EOL> self . save_history ( ) <EOL> return rc <EOL> def read_history ( self ) : <EOL> if hasReadline and self . _history_path : <EOL> try : <EOL> readline . read_history_file ( self . _history_path ) <EOL> except : <EOL> pass <EOL> def save_history ( self ) : <EOL> if hasReadline and self . _history_path : <EOL> readline . set_history_length ( <NUM_LIT:1000> ) <EOL> try : <EOL> readline . write_history_file ( self . _history_path ) <EOL> except : <EOL> pass <EOL> def completedefault ( self , text , line , begin , end ) : <EOL> line = line . strip ( ) . lower ( ) <EOL> tables = self . db . tables . keys ( ) <EOL> views = self . db . views . keys ( ) <EOL> if ( line . endswith ( '<STR_LIT>' ) or line . endswith ( '<STR_LIT>' ) <EOL> or line . endswith ( '<STR_LIT>' ) ) : <EOL> candidates = tables + views <EOL> elif line . endswith ( '<STR_LIT>' ) : <EOL> candidates = tables <EOL> elif line . endswith ( '<STR_LIT>' ) : <EOL> candidates = views <EOL> else : <EOL> candidates = tables + views + self . sqlkeywords <EOL> text = text . lower ( ) <EOL> return [ a + '<STR_LIT:U+0020>' for a in candidates if a . startswith ( text ) ] <EOL> def completenames ( self , text , line , begin , end ) : <EOL> cmds = [ '<STR_LIT:.>' + a [ <NUM_LIT:4> : ] for a in self . get_names ( ) <EOL> if a . startswith ( '<STR_LIT>' ) ] <EOL> if line . endswith ( '<STR_LIT>' ) : <EOL> starters = cmds <EOL> else : <EOL> starters = cmds + self . sqlstarters <EOL> cmd , arg , line = self . parseline ( text ) <EOL> if not cmd : <EOL> return starters <EOL> text = text . lower ( ) <EOL> return [ a + '<STR_LIT:U+0020>' for a in starters if a . startswith ( text ) ] <EOL> def print_topics ( self , header , cmds , cmdlen , maxcol ) : <EOL> if cmds == [ '<STR_LIT>' ] : <EOL> return <EOL> if cmds : <EOL> self . stdout . write ( "<STR_LIT>" % str ( header ) ) <EOL> if self . ruler : <EOL> self . stdout . write ( "<STR_LIT>" % str ( self . ruler * len ( header ) ) ) <EOL> newcmds = [ ] <EOL> for cmd in cmds : <EOL> if cmd . startswith ( '<STR_LIT:_>' ) : <EOL> newcmds . append ( '<STR_LIT:.>' + cmd [ <NUM_LIT:1> : ] ) <EOL> else : <EOL> newcmds . append ( cmd ) <EOL> self . columnize ( newcmds , maxcol - <NUM_LIT:1> ) <EOL> self . stdout . write ( "<STR_LIT:\n>" ) <EOL> def complete__yesno ( self , text , * ignored ) : <EOL> return [ x for x in itertools . chain ( self . yes_args , self . no_args ) <EOL> if x . startswith ( text ) ] <EOL> def complete__noop ( self , * args ) : <EOL> return [ ] <EOL> schemaBits = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def do__show ( self , arg ) : <EOL> self . db . loadSchema ( ) <EOL> if arg in self . schemaBits : <EOL> d = getattr ( self . db , arg ) <EOL> print '<STR_LIT:\n>' . join ( sorted ( d . keys ( ) ) ) <EOL> else : <EOL> print '<STR_LIT>' , arg <EOL> return False <EOL> def help__show ( self ) : <EOL> print """<STR_LIT>""" % '<STR_LIT:U+0020>' . join ( '<STR_LIT>' % x for x in self . schemaBits ) <EOL> def complete__show ( self , text , * ignored ) : <EOL> return [ x for x in self . schemaBits if x . startswith ( text ) ] <EOL> def do__headers ( self , arg ) : <EOL> if arg in self . yes_args : <EOL> self . show_headers = True <EOL> elif arg in self . no_args : <EOL> self . show_headers = False <EOL> else : <EOL> print '<STR_LIT>' , arg <EOL> return False <EOL> def help__headers ( self ) : <EOL> print """<STR_LIT>""" <EOL> do__head = do__headers <EOL> help__head = help__headers <EOL> complete__headers = complete__yesno <EOL> complete__head = complete__yesno <EOL> def do__pager ( self , arg ) : <EOL> if arg in self . yes_args : <EOL> self . use_pager = True <EOL> elif arg in self . no_args : <EOL> self . use_pager = False <EOL> else : <EOL> print '<STR_LIT>' , arg <EOL> return False <EOL> def help__pager ( self ) : <EOL> print """<STR_LIT>""" <EOL> complete__pager = complete__yesno <EOL> def do__stats ( self , arg ) : <EOL> if arg in self . yes_args : <EOL> self . show_stats = True <EOL> elif arg in self . no_args : <EOL> self . show_stats = False <EOL> else : <EOL> print '<STR_LIT>' , arg <EOL> return False <EOL> def help__stats ( self ) : <EOL> print """<STR_LIT>""" <EOL> complete__stats = complete__yesno <EOL> def set_mode ( self , mode ) : <EOL> self . format = getattr ( self , '<STR_LIT>' + mode ) <EOL> modes = ( '<STR_LIT>' , '<STR_LIT:list>' ) <EOL> def do__mode ( self , arg ) : <EOL> choices = [ x for x in self . modes if x . startswith ( arg ) ] <EOL> if len ( choices ) != <NUM_LIT:1> : <EOL> print '<STR_LIT>' , arg <EOL> return False <EOL> self . set_mode ( choices [ <NUM_LIT:0> ] ) <EOL> return False <EOL> def help__mode ( self ) : <EOL> print """<STR_LIT>""" % '<STR_LIT:U+0020>' . join ( '<STR_LIT>' % x for x in self . modes ) <EOL> def complete__mode ( self , text , * ignored ) : <EOL> return [ x for x in self . modes if x . startswith ( text ) ] <EOL> do__head = do__headers <EOL> help__head = help__headers <EOL> complete__head = complete__headers <EOL> def do__width ( self , arg ) : <EOL> if arg == '<STR_LIT>' : <EOL> self . auto_width = True <EOL> elif arg == '<STR_LIT>' : <EOL> self . auto_width = False <EOL> elif '<STR_LIT:=>' in arg : <EOL> col , width = arg . split ( '<STR_LIT:=>' , <NUM_LIT:1> ) <EOL> col = col . strip ( ) <EOL> width = width . strip ( ) <EOL> try : <EOL> col = int ( col ) <EOL> except ValueError : <EOL> print '<STR_LIT>' % col <EOL> return False <EOL> try : <EOL> width = int ( width ) <EOL> except ValueError : <EOL> print '<STR_LIT>' % width <EOL> return False <EOL> if col < <NUM_LIT:0> : <EOL> print '<STR_LIT>' % col <EOL> self . manual_widths [ col - <NUM_LIT:1> ] = width <EOL> else : <EOL> new_widths = { } <EOL> for col , width in enumerate ( arg . split ( ) ) : <EOL> width = width . strip ( ) <EOL> try : <EOL> width = int ( width ) <EOL> except ValueError : <EOL> print '<STR_LIT>' % width <EOL> return False <EOL> new_widths [ col ] = width <EOL> self . manual_widths . update ( new_widths ) <EOL> def help__width ( self ) : <EOL> print """<STR_LIT>""" <EOL> complete__width = complete__noop <EOL> def do__quit ( self , arg ) : <EOL> return True <EOL> def help__quit ( self ) : <EOL> print """<STR_LIT>""" <EOL> complete__quit = complete__noop <EOL> def do__reset ( self , arg ) : <EOL> sys . stdout . write ( '<STR_LIT>' ) <EOL> sys . stdout . flush ( ) <EOL> def help__reset ( self ) : <EOL> print """<STR_LIT>""" <EOL> complete__reset = complete__noop <EOL> def do__help ( self , arg ) : <EOL> if arg . startswith ( '<STR_LIT:.>' ) : <EOL> arg = '<STR_LIT:_>' + arg [ <NUM_LIT:1> : ] <EOL> return cmd . Cmd . do_help ( self , arg ) <EOL> def help__help ( self ) : <EOL> print """<STR_LIT>""" <EOL> complete__help = cmd . Cmd . complete_help <EOL> def shell ( db ) : <EOL> '<STR_LIT>' <EOL> shell = DbShell ( db ) <EOL> return shell . cmdloop ( ) </s>
