<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> __docformat__ = "<STR_LIT>" <EOL> import re <EOL> import time <EOL> import os <EOL> import sys <EOL> from datetime import date , datetime <EOL> from collections import namedtuple <EOL> from grizzled . exception import ExceptionWithMessage <EOL> from grizzled . decorators import abstract <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> class DBError ( ExceptionWithMessage ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class Error ( DBError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class Warning ( DBError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> TableMetadata = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> IndexMetadata = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:description>' ] ) <EOL> RDBMSMetadata = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:version>' ] ) <EOL> class Cursor ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , cursor , driver ) : <EOL> """<STR_LIT>""" <EOL> self . __cursor = cursor <EOL> self . __driver = driver <EOL> self . __description = None <EOL> self . __rowcount = - <NUM_LIT:1> <EOL> def __get_description ( self ) : <EOL> return self . __description <EOL> description = property ( __get_description , <EOL> doc = '<STR_LIT>' ) <EOL> def __get_rowcount ( self ) : <EOL> return self . __rowcount <EOL> rowcount = property ( __get_rowcount , <EOL> doc = '<STR_LIT>' ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> return self . __cursor . close ( ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> def execute ( self , statement , parameters = None ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> if parameters : <EOL> result = self . __cursor . execute ( statement , parameters ) <EOL> else : <EOL> result = self . __cursor . execute ( statement ) <EOL> try : <EOL> self . __rowcount = self . __cursor . rowcount <EOL> except AttributeError : <EOL> self . __rowcount = - <NUM_LIT:1> <EOL> self . __description = self . __cursor . description <EOL> return result <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> except : <EOL> raise Error ( sys . exc_info ( ) [ <NUM_LIT:1> ] ) <EOL> def executemany ( self , statement , * parameters ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> result = self . __cursor . executemany ( statement , * parameters ) <EOL> self . __rowcount = self . __cursor . rowcount <EOL> self . __description = self . __cursor . description <EOL> return result <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> executeMany = executemany <EOL> def fetchone ( self ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> return self . __cursor . fetchone ( ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> def fetchall ( self ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> return self . __cursor . fetchall ( ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> fetchAll = fetchall <EOL> def fetchmany ( self , n ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> self . __cursor . fetchmany ( n ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> fetchMany = fetchmany <EOL> def get_rdbms_metadata ( self ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> return self . __driver . get_rdbms_metadata ( self . __cursor ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> def get_table_metadata ( self , table ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> return self . __driver . get_table_metadata ( table , self . __cursor ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> def get_index_metadata ( self , table ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> return self . __driver . get_index_metadata ( table , self . __cursor ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> def get_tables ( self ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> return self . __driver . get_tables ( self . __cursor ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> class DB ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , db , driver ) : <EOL> """<STR_LIT>""" <EOL> self . __db = db <EOL> self . __driver = driver <EOL> dbi = driver . get_import ( ) <EOL> for attr in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> try : <EOL> exec '<STR_LIT>' % ( attr , attr ) <EOL> except AttributeError : <EOL> exec '<STR_LIT>' % attr <EOL> def paramstyle ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __driver . get_import ( ) . paramstyle <EOL> def Binary ( self , string ) : <EOL> """<STR_LIT>""" <EOL> return self . __driver . get_import ( ) . Binary ( string ) <EOL> def Date ( self , year , month , day ) : <EOL> """<STR_LIT>""" <EOL> return self . __driver . get_import ( ) . Date ( year , month , day ) <EOL> def DateFromTicks ( self , secs ) : <EOL> """<STR_LIT>""" <EOL> date = date . fromtimestamp ( secs ) <EOL> return self . __driver . get_import ( ) . Date ( date . year , date . month , date . day ) <EOL> def Time ( self , hour , minute , second ) : <EOL> """<STR_LIT>""" <EOL> dt = datetime . fromtimestamp ( secs ) <EOL> return self . __driver . get_import ( ) . Time ( dt . hour , dt . minute , dt . second ) <EOL> def TimeFromTicks ( self , secs ) : <EOL> """<STR_LIT>""" <EOL> dt = datetime . fromtimestamp ( secs ) <EOL> return self . __driver . get_import ( ) . Time ( dt . hour , dt . minute , dt . second ) <EOL> def Timestamp ( self , year , month , day , hour , minute , second ) : <EOL> """<STR_LIT>""" <EOL> return self . __driver . get_import ( ) . Timestamp ( year , month , day , <EOL> hour , minute , second ) <EOL> def TimestampFromTicks ( self , secs ) : <EOL> """<STR_LIT>""" <EOL> dt = datetime . now ( ) <EOL> return self . __driver . get_import ( ) . Timestamp ( dt . year , dt . month , dt . day , <EOL> dt . hour , dt . minute , dt . second ) <EOL> def cursor ( self ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> return Cursor ( self . __db . cursor ( ) , self . __driver ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> def commit ( self ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> self . __db . commit ( ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> def rollback ( self ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> self . __db . rollback ( ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . __driver . get_import ( ) <EOL> try : <EOL> self . __db . close ( ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> class DBDriver ( object ) : <EOL> """<STR_LIT>""" <EOL> @ abstract <EOL> def get_import ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def __display_name ( self ) : <EOL> return self . get_display_name ( ) <EOL> @ abstract <EOL> def get_display_name ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> display_name = property ( __display_name , <EOL> doc = '<STR_LIT>' ) <EOL> def connect ( self , <EOL> host = '<STR_LIT:localhost>' , <EOL> port = None , <EOL> user = None , <EOL> password = '<STR_LIT>' , <EOL> database = None ) : <EOL> """<STR_LIT>""" <EOL> dbi = self . get_import ( ) <EOL> try : <EOL> self . __db = self . do_connect ( host = host , <EOL> port = port , <EOL> user = user , <EOL> password = password , <EOL> database = database ) <EOL> return DB ( self . __db , self ) <EOL> except dbi . Warning , val : <EOL> raise Warning ( val ) <EOL> except dbi . Error , val : <EOL> raise Error ( val ) <EOL> @ abstract <EOL> def do_connect ( self , <EOL> host = '<STR_LIT:localhost>' , <EOL> port = None , <EOL> user = '<STR_LIT>' , <EOL> password = '<STR_LIT>' , <EOL> database = '<STR_LIT:default>' ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def get_rdbms_metadata ( self , cursor ) : <EOL> """<STR_LIT>""" <EOL> return RDBMSMetadata ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def get_index_metadata ( self , table , cursor ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def get_table_metadata ( self , table , cursor ) : <EOL> """<STR_LIT>""" <EOL> self . _ensure_valid_table ( cursor , table ) <EOL> dbi = self . get_import ( ) <EOL> cursor . execute ( '<STR_LIT>' % table ) <EOL> result = [ ] <EOL> for col in cursor . description : <EOL> name = col [ <NUM_LIT:0> ] <EOL> type = col [ <NUM_LIT:1> ] <EOL> size = col [ <NUM_LIT:2> ] <EOL> internalSize = col [ <NUM_LIT:3> ] <EOL> precision = col [ <NUM_LIT:4> ] <EOL> scale = col [ <NUM_LIT:5> ] <EOL> nullable = col [ <NUM_LIT:6> ] <EOL> sType = None <EOL> try : <EOL> if type == dbi . BINARY : <EOL> stype = '<STR_LIT>' <EOL> elif type == dbi . DATETIME : <EOL> stype = '<STR_LIT>' <EOL> elif type == dbi . NUMBER : <EOL> stype = '<STR_LIT>' <EOL> elif type == dbi . STRING : <EOL> sz = internalSize <EOL> if sz == None : <EOL> sz = size <EOL> elif sz <= <NUM_LIT:0> : <EOL> sz = size <EOL> if sz == <NUM_LIT:1> : <EOL> stype = '<STR_LIT>' <EOL> else : <EOL> stype = '<STR_LIT>' <EOL> size = sz <EOL> elif type == dbi . ROWID : <EOL> stype = '<STR_LIT:id>' <EOL> except AttributeError : <EOL> stype = None <EOL> if not sType : <EOL> stype = '<STR_LIT>' % str ( type ) <EOL> data = TableMetadata ( name , stype , size , precision , scale , nullable ) <EOL> result += [ data ] <EOL> return result <EOL> def get_tables ( self , cursor ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def _ensure_valid_table ( self , cursor , table_name ) : <EOL> """<STR_LIT>""" <EOL> if not self . _is_valid_table ( cursor , table_name ) : <EOL> raise Error , '<STR_LIT>' % table_name <EOL> def _is_valid_table ( self , cursor , table_name ) : <EOL> """<STR_LIT>""" <EOL> tables = self . get_tables ( cursor ) <EOL> return table_name in tables </s>
