<s> import logging <EOL> try : <EOL> try : <EOL> from pysqlite2 import dbapi2 as Database <EOL> except ImportError , e1 : <EOL> from sqlite3 import dbapi2 as Database <EOL> except ImportError , exc : <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> raise ImproperlyConfigured ( "<STR_LIT>" % exc ) <EOL> from librdbms . server . rdbms_base_lib import BaseRDBMSDataTable , BaseRDBMSResult , BaseRDMSClient <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class DataTable ( BaseRDBMSDataTable ) : pass <EOL> class Result ( BaseRDBMSResult ) : pass <EOL> class SQLiteClient ( BaseRDMSClient ) : <EOL> """<STR_LIT>""" <EOL> data_table_cls = DataTable <EOL> result_cls = Result <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( SQLiteClient , self ) . __init__ ( * args , ** kwargs ) <EOL> self . connection = Database . connect ( ** self . _conn_params ) <EOL> @ property <EOL> def _conn_params ( self ) : <EOL> params = { <EOL> '<STR_LIT>' : self . query_server [ '<STR_LIT:name>' ] , <EOL> '<STR_LIT>' : Database . PARSE_DECLTYPES | Database . PARSE_COLNAMES , <EOL> } <EOL> if self . query_server [ '<STR_LIT>' ] : <EOL> params . update ( self . query_server [ '<STR_LIT>' ] ) <EOL> params [ '<STR_LIT>' ] = False <EOL> return params <EOL> def use ( self , database ) : <EOL> pass <EOL> def execute_statement ( self , statement ) : <EOL> cursor = self . connection . cursor ( ) <EOL> cursor . execute ( statement ) <EOL> self . connection . commit ( ) <EOL> if cursor . description : <EOL> columns = [ column [ <NUM_LIT:0> ] for column in cursor . description ] <EOL> else : <EOL> columns = [ ] <EOL> return self . data_table_cls ( cursor , columns ) <EOL> def get_databases ( self ) : <EOL> return [ self . _conn_params [ '<STR_LIT>' ] ] <EOL> def get_tables ( self , database , table_names = [ ] ) : <EOL> cursor = self . connection . cursor ( ) <EOL> query = "<STR_LIT>" <EOL> if table_names : <EOL> clause = '<STR_LIT>' . join ( [ "<STR_LIT>" % { '<STR_LIT>' : table } for table in table_names ] ) <EOL> query += '<STR_LIT>' % clause <EOL> cursor . execute ( query ) <EOL> self . connection . commit ( ) <EOL> return [ row [ <NUM_LIT:0> ] for row in cursor . fetchall ( ) ] <EOL> def get_columns ( self , database , table , names_only = True ) : <EOL> cursor = self . connection . cursor ( ) <EOL> cursor . execute ( "<STR_LIT>" % table ) <EOL> self . connection . commit ( ) <EOL> if names_only : <EOL> columns = [ row [ <NUM_LIT:1> ] for row in cursor . fetchall ( ) ] <EOL> else : <EOL> columns = [ dict ( name = row [ <NUM_LIT:1> ] , type = row [ <NUM_LIT:2> ] , comment = '<STR_LIT>' ) for row in cursor . fetchall ( ) ] <EOL> return columns <EOL> def get_sample_data ( self , database , table , column = None , limit = <NUM_LIT:100> ) : <EOL> column = '<STR_LIT>' % column if column else '<STR_LIT:*>' <EOL> statement = '<STR_LIT>' % ( column , table , limit ) <EOL> return self . execute_statement ( statement ) </s>
