<s> import sys <EOL> import os <EOL> import sqlite3 <EOL> import re <EOL> import logging <EOL> import time <EOL> import string <EOL> from util import Identity , unicode_safe_str <EOL> import apsw_connection <EOL> import qasino_table <EOL> import apsw <EOL> from twisted . enterprise import adbapi <EOL> from twisted . internet import reactor <EOL> import thread <EOL> class CreateTableException ( Exception ) : <EOL> pass <EOL> class SqlConnections ( object ) : <EOL> """<STR_LIT>""" <EOL> WRITER_INTERACTION = <NUM_LIT:0> <EOL> READER_INTERACTION = <NUM_LIT:1> <EOL> def __init__ ( self , filename , data_manager , archive_db_dir , thread_id , static_filename ) : <EOL> self . data_manager = data_manager <EOL> self . tables = { } <EOL> self . connections = { } <EOL> self . stats = { } <EOL> self . archive_db_dir = archive_db_dir <EOL> self . main_thread = thread_id <EOL> self . open_new_db ( filename , static_filename ) <EOL> def open_new_db ( self , filename , static_filename ) : <EOL> self . filename = filename <EOL> self . static_filename = static_filename <EOL> adbapi . connectionFactory = apsw_connection . ApswConnection <EOL> def attach_static_db ( conn ) : <EOL> if static_filename != None and os . path . exists ( static_filename ) : <EOL> cur = conn . cursor ( ) <EOL> try : <EOL> result = cur . execute ( "<STR_LIT>" % static_filename ) <EOL> except Exception as e : <EOL> logging . info ( "<STR_LIT>" , static_filename , str ( e ) ) <EOL> self . writer_dbpool = adbapi . ConnectionPool ( '<STR_LIT>' , filename , <EOL> cp_openfun = attach_static_db , <EOL> cp_min = <NUM_LIT:1> , cp_max = <NUM_LIT:1> ) <EOL> self . writer_dbpool . connectionFactory = apsw_connection . ApswConnection <EOL> self . reader_dbpool = adbapi . ConnectionPool ( '<STR_LIT>' , filename , <EOL> cp_openfun = attach_static_db ) <EOL> self . reader_dbpool . connectionFactory = apsw_connection . ApswConnection <EOL> self . reader_dbpool . runInteraction ( self . preload_tables_list ) <EOL> def __del__ ( self ) : <EOL> if self . main_thread == thread . get_ident ( ) : <EOL> SqlConnections . shutdown ( self . writer_dbpool , self . filename , self . archive_db_dir ) <EOL> SqlConnections . shutdown ( self . reader_dbpool , self . filename , self . archive_db_dir ) <EOL> else : <EOL> reactor . callLater ( <NUM_LIT:0.1> , SqlConnections . shutdown , self . writer_dbpool , self . filename , self . archive_db_dir ) <EOL> reactor . callLater ( <NUM_LIT:0.1> , SqlConnections . shutdown , self . reader_dbpool , self . filename , self . archive_db_dir ) <EOL> @ staticmethod <EOL> def shutdown ( dbpool , filename , archive_db_dir ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> dbpool . close ( ) <EOL> except Exception as e : <EOL> logging . info ( "<STR_LIT>" , e ) <EOL> if filename != '<STR_LIT>' and os . path . exists ( filename ) : <EOL> if archive_db_dir : <EOL> logging . info ( "<STR_LIT>" , filename , archive_db_dir ) <EOL> try : <EOL> os . renames ( filename , archive_db_dir ) <EOL> except Exception as e : <EOL> logging . info ( "<STR_LIT>" , e ) <EOL> else : <EOL> logging . info ( "<STR_LIT>" , filename ) <EOL> try : <EOL> os . remove ( filename ) <EOL> except Exception as e : <EOL> logging . info ( "<STR_LIT>" , e ) <EOL> def do_select ( self , txn , sql ) : <EOL> self . stats [ '<STR_LIT>' ] = self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> try : <EOL> retval = txn . execute ( sql ) <EOL> except Exception as e : <EOL> self . stats [ '<STR_LIT>' ] = self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> return { "<STR_LIT>" : <NUM_LIT:1> , "<STR_LIT>" : str ( e ) } <EOL> max_widths = { } <EOL> column_names = [ ] <EOL> try : <EOL> for column_index , column_name in enumerate ( txn . getdescription ( ) ) : <EOL> if not max_widths . has_key ( str ( column_index ) ) or max_widths [ str ( column_index ) ] < len ( column_name [ <NUM_LIT:0> ] ) : <EOL> max_widths [ str ( column_index ) ] = len ( column_name [ <NUM_LIT:0> ] ) <EOL> column_names . append ( column_name [ <NUM_LIT:0> ] ) <EOL> except apsw . ExecutionCompleteError as e : <EOL> data = { "<STR_LIT>" : [ "<STR_LIT>" ] , "<STR_LIT>" : [ ] } <EOL> return { "<STR_LIT>" : <NUM_LIT:0> , "<STR_LIT>" : '<STR_LIT>' , "<STR_LIT:data>" : data , "<STR_LIT>" : { "<STR_LIT:0>" : <NUM_LIT> } } <EOL> saved_rows = [ ] <EOL> nr_rows = <NUM_LIT:0> <EOL> try : <EOL> for row in txn . fetchall ( ) : <EOL> saved_row = [ ] <EOL> for column_index , cell in enumerate ( row ) : <EOL> cell = unicode_safe_str ( cell ) <EOL> length = len ( cell ) <EOL> if not max_widths . has_key ( str ( column_index ) ) or max_widths [ str ( column_index ) ] < length : <EOL> max_widths [ str ( column_index ) ] = length <EOL> saved_row . append ( cell ) <EOL> saved_rows . append ( saved_row ) <EOL> except ValueError as e : <EOL> self . stats [ '<STR_LIT>' ] = self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> return { "<STR_LIT>" : <NUM_LIT:1> , "<STR_LIT>" : str ( e ) } <EOL> self . stats [ '<STR_LIT>' ] = self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> data = { "<STR_LIT>" : column_names , <EOL> "<STR_LIT>" : saved_rows } <EOL> return { "<STR_LIT>" : <NUM_LIT:0> , "<STR_LIT>" : '<STR_LIT>' , "<STR_LIT:data>" : data , "<STR_LIT>" : max_widths } <EOL> def get_schema ( self , txn , table ) : <EOL> schema = [ ] <EOL> try : <EOL> txn . execute ( "<STR_LIT>" % table ) <EOL> except Exception as e : <EOL> logging . info ( "<STR_LIT>" , table ) <EOL> return schema <EOL> for row in txn . fetchall ( ) : <EOL> schema . append ( [ str ( row [ <NUM_LIT:1> ] ) , str ( row [ <NUM_LIT:2> ] ) ] ) <EOL> return schema <EOL> def do_sql ( self , txn , sql ) : <EOL> """<STR_LIT>""" <EOL> return txn . execute ( sql ) <EOL> def do_desc ( self , txn , tablename ) : <EOL> schema = self . get_schema ( txn , tablename ) <EOL> if schema == None or len ( schema ) <= <NUM_LIT:0> : <EOL> return ( <NUM_LIT:1> , "<STR_LIT>" % ( tablename , ) , None ) <EOL> table = { "<STR_LIT>" : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> "<STR_LIT>" : schema } <EOL> return ( <NUM_LIT:0> , '<STR_LIT>' , table ) <EOL> def do_update_table ( self , txn , table , identity ) : <EOL> self . stats [ '<STR_LIT>' ] = self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> key_cols_str = table . get_property ( '<STR_LIT>' ) <EOL> column_names = table . get_column_names ( ) <EOL> if key_cols_str == None : <EOL> key_column_names = column_names <EOL> else : <EOL> key_column_names = string . split ( key_cols_str , '<STR_LIT:;>' ) <EOL> tablename = table . get_tablename ( ) <EOL> index_name = tablename + "<STR_LIT>" + '<STR_LIT:_>' . join ( key_column_names ) <EOL> create_index_sql = "<STR_LIT>" % ( index_name , tablename , '<STR_LIT:U+002CU+0020>' . join ( key_column_names ) ) <EOL> try : <EOL> txn . execute ( create_index_sql ) <EOL> except Exception as e : <EOL> logging . info ( "<STR_LIT>" , str ( e ) , create_index_sql ) <EOL> self . stats [ '<STR_LIT>' ] = self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> return <EOL> sql = "<STR_LIT>" % ( tablename , '<STR_LIT:U+002CU+0020>' . join ( column_names ) , '<STR_LIT:U+002CU+0020>' . join ( [ '<STR_LIT:?>' for x in column_names ] ) ) <EOL> rowcount = <NUM_LIT:0> <EOL> for row in table . get_rows ( ) : <EOL> txn . execute ( sql , row ) <EOL> rowcount += txn . getconnection ( ) . changes ( ) <EOL> self . stats [ '<STR_LIT>' ] = self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> return rowcount <EOL> def do_insert_table ( self , txn , table ) : <EOL> """<STR_LIT>""" <EOL> self . stats [ '<STR_LIT>' ] = self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> rowcount = <NUM_LIT:0> <EOL> base_query = "<STR_LIT>" % ( table . get_tablename ( ) , "<STR_LIT:U+002C>" . join ( table . get_column_names ( ) ) ) <EOL> nr_columns = len ( table . get_column_names ( ) ) <EOL> max_nr_values = <NUM_LIT> <EOL> nr_values = <NUM_LIT:0> <EOL> bind_values = [ ] <EOL> one_row = '<STR_LIT>' . format ( '<STR_LIT:U+002C>' . join ( [ '<STR_LIT:?>' ] * nr_columns ) ) <EOL> sql = base_query <EOL> for row_index , row in enumerate ( table . get_rows ( ) ) : <EOL> if nr_values + nr_columns > max_nr_values : <EOL> sql += '<STR_LIT:U+002C>' . join ( [ one_row ] * ( nr_values / nr_columns ) ) <EOL> txn . execute ( sql , bind_values ) <EOL> bind_values = [ ] <EOL> nr_values = <NUM_LIT:0> <EOL> sql = base_query <EOL> nr_values += nr_columns <EOL> for value in row : <EOL> bind_values . append ( value ) <EOL> rowcount += <NUM_LIT:1> <EOL> if rowcount > <NUM_LIT:0> : <EOL> sql += '<STR_LIT:U+002C>' . join ( [ one_row ] * ( nr_values / nr_columns ) ) <EOL> txn . execute ( sql , bind_values ) <EOL> self . stats [ '<STR_LIT>' ] = self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> return rowcount <EOL> def run_interaction ( self , interaction_type , callback , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if interaction_type == SqlConnections . WRITER_INTERACTION : <EOL> return self . writer_dbpool . runInteraction ( callback , * args , ** kwargs ) <EOL> elif interaction_type == SqlConnections . READER_INTERACTION : <EOL> return self . reader_dbpool . runInteraction ( callback , * args , ** kwargs ) <EOL> return <EOL> def insert_info_table ( self , txn , db_generation_number , generation_start_epoch , generation_duration_s ) : <EOL> """<STR_LIT>""" <EOL> table = qasino_table . QasinoTable ( "<STR_LIT>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_row ( [ str ( db_generation_number ) , generation_duration_s , generation_start_epoch ] ) <EOL> return self . add_table_data ( txn , table , Identity . get_identity ( ) ) <EOL> def insert_sql_stats_table ( self , txn , sql_backend_reader ) : <EOL> """<STR_LIT>""" <EOL> table = qasino_table . QasinoTable ( "<STR_LIT>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_row ( [ sql_backend_reader . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) , <EOL> sql_backend_reader . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) , <EOL> sql_backend_reader . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) ] ) <EOL> return self . add_table_data ( txn , table , Identity . get_identity ( ) ) <EOL> def insert_update_stats_table ( self , txn ) : <EOL> """<STR_LIT>""" <EOL> table = qasino_table . QasinoTable ( "<STR_LIT>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_row ( [ self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) , <EOL> self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) , <EOL> self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) , <EOL> self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) , <EOL> self . stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) ] ) <EOL> return self . add_table_data ( txn , table , Identity . get_identity ( ) ) <EOL> def add_tables_table_rows ( self , table ) : <EOL> for tablename , table_data in self . tables . items ( ) : <EOL> table . add_row ( [ tablename , <EOL> str ( table_data [ "<STR_LIT>" ] ) , <EOL> str ( table_data [ "<STR_LIT>" ] ) , <EOL> table_data [ "<STR_LIT>" ] , <EOL> <NUM_LIT:0> if self . static_filename != None else <NUM_LIT:1> ] <EOL> ) <EOL> def preload_tables_list ( self , txn ) : <EOL> """<STR_LIT>""" <EOL> self . do_sql ( txn , "<STR_LIT>" ) <EOL> rows = txn . fetchall ( ) <EOL> for row in rows : <EOL> tablename = row [ <NUM_LIT:0> ] <EOL> try : <EOL> self . do_sql ( txn , "<STR_LIT>" % tablename ) <EOL> count = txn . fetchall ( ) <EOL> self . update_table_stats ( tablename , count [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] if count [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] != None and count [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] > <NUM_LIT:0> else - <NUM_LIT:1> ) <EOL> except : <EOL> logging . info ( "<STR_LIT>" , tablename , str ( e ) ) <EOL> pass <EOL> def insert_connections_table ( self , txn ) : <EOL> """<STR_LIT>""" <EOL> table = qasino_table . QasinoTable ( "<STR_LIT>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> for connection , connection_data in self . connections . items ( ) : <EOL> table . add_row ( [ connection , <EOL> str ( len ( connection_data [ "<STR_LIT>" ] ) ) , <EOL> connection_data [ "<STR_LIT>" ] ] ) <EOL> return self . add_table_data ( txn , table , Identity . get_identity ( ) ) <EOL> def add_views ( self , txn , views ) : <EOL> """<STR_LIT>""" <EOL> for viewname , viewdata in views . iteritems ( ) : <EOL> logging . info ( "<STR_LIT>" , viewname ) <EOL> try : <EOL> self . do_sql ( txn , viewdata [ '<STR_LIT>' ] ) <EOL> viewdata [ '<STR_LIT>' ] = True <EOL> viewdata [ '<STR_LIT:error>' ] = '<STR_LIT>' <EOL> except Exception as e : <EOL> logging . info ( "<STR_LIT>" , viewname , str ( e ) ) <EOL> viewdata [ '<STR_LIT>' ] = False <EOL> viewdata [ '<STR_LIT:error>' ] = str ( e ) <EOL> def insert_views_table ( self , txn , views ) : <EOL> """<STR_LIT>""" <EOL> table = qasino_table . QasinoTable ( "<STR_LIT>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT:int>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> table . add_column ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> for viewname , viewdata in views . iteritems ( ) : <EOL> table . add_row ( [ viewname , str ( int ( viewdata [ '<STR_LIT>' ] ) ) , str ( viewdata [ '<STR_LIT:error>' ] ) , viewdata [ '<STR_LIT>' ] ] ) <EOL> return self . add_table_data ( txn , table , Identity . get_identity ( ) ) <EOL> def update_table_stats ( self , tablename , nr_rows , identity = Identity . get_identity ( ) , now = time . time ( ) , sum = False ) : <EOL> if tablename not in self . tables : <EOL> self . tables [ tablename ] = { "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : nr_rows , <EOL> "<STR_LIT>" : now } <EOL> else : <EOL> if sum : <EOL> self . tables [ tablename ] [ "<STR_LIT>" ] += nr_rows <EOL> self . tables [ tablename ] [ "<STR_LIT>" ] += <NUM_LIT:1> <EOL> self . tables [ tablename ] [ "<STR_LIT>" ] = now <EOL> if identity not in self . connections : <EOL> self . connections [ identity ] = { '<STR_LIT>' : { tablename : nr_rows } , <EOL> '<STR_LIT>' : now } <EOL> else : <EOL> self . connections [ identity ] [ "<STR_LIT>" ] = now <EOL> if sum and tablename in self . connections [ identity ] [ "<STR_LIT>" ] : <EOL> self . connections [ identity ] [ "<STR_LIT>" ] [ tablename ] += nr_rows <EOL> else : <EOL> self . connections [ identity ] [ "<STR_LIT>" ] [ tablename ] = nr_rows <EOL> def async_add_table_data ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . writer_dbpool . runInteraction ( self . add_table_data , * args , ** kwargs ) <EOL> def add_table_data ( self , txn , table , identity ) : <EOL> """<STR_LIT>""" <EOL> start_time = time . time ( ) <EOL> table . init_retry ( <NUM_LIT:5> ) <EOL> tablename = table . get_tablename ( ) <EOL> if tablename is None : <EOL> logging . info ( "<STR_LIT>" , identity ) <EOL> return <EOL> update = table . get_property ( '<STR_LIT>' ) <EOL> static = table . get_property ( '<STR_LIT>' ) <EOL> persist = table . get_property ( '<STR_LIT>' ) <EOL> self . data_manager . check_save_table ( table , identity ) <EOL> try : <EOL> if not update and not static and self . connections [ identity ] [ "<STR_LIT>" ] [ tablename ] > <NUM_LIT:0> : <EOL> logging . info ( "<STR_LIT>" , identity , tablename ) <EOL> return <EOL> except KeyError : <EOL> pass <EOL> try : <EOL> txn . execute ( "<STR_LIT>" ) <EOL> create_sql = "<STR_LIT>" % tablename <EOL> columns = [ ] <EOL> for column_name , column_type in table . zip_columns ( ) : <EOL> columns . append ( "<STR_LIT>" % ( column_name , column_type ) ) <EOL> create_sql += "<STR_LIT>" . join ( columns ) <EOL> create_sql += "<STR_LIT:)>" <EOL> try : <EOL> result = self . do_sql ( txn , create_sql ) <EOL> except Exception as e : <EOL> errorstr = str ( e ) <EOL> if "<STR_LIT>" in errorstr : <EOL> raise CreateTableException ( "<STR_LIT>" % ( tablename , identity , errorstr , ) ) <EOL> else : <EOL> raise e <EOL> schema = self . get_schema ( txn , tablename ) <EOL> self . data_manager . table_merger . merge_table ( txn , table , schema , self ) <EOL> if static : <EOL> txn . execute ( "<STR_LIT>" % tablename ) <EOL> rowcount = self . do_insert_table ( txn , table ) <EOL> else : <EOL> if update : <EOL> rowcount = self . do_update_table ( txn , table , identity ) <EOL> else : <EOL> rowcount = self . do_insert_table ( txn , table ) <EOL> txn . execute ( "<STR_LIT>" ) <EOL> except apsw . BusyError as e : <EOL> logging . info ( "<STR_LIT>" , tablename , str ( e ) , identity ) <EOL> try : <EOL> txn . execute ( "<STR_LIT>" ) <EOL> except : <EOL> pass <EOL> if table . test_retry ( ) : <EOL> logging . info ( "<STR_LIT>" , tablename ) <EOL> return <NUM_LIT:0> <EOL> self . async_add_table_data ( table , identity ) <EOL> return <NUM_LIT:0> <EOL> except CreateTableException as e : <EOL> try : <EOL> txn . execute ( "<STR_LIT>" ) <EOL> except : <EOL> pass <EOL> if table . test_retry ( ) : <EOL> logging . info ( "<STR_LIT>" , tablename ) <EOL> return <NUM_LIT:0> <EOL> self . async_add_table_data ( table , identity ) <EOL> return <NUM_LIT:0> <EOL> except Exception as e : <EOL> logging . info ( "<STR_LIT>" , tablename , e ) <EOL> try : <EOL> txn . execute ( "<STR_LIT>" ) <EOL> except : <EOL> pass <EOL> return <NUM_LIT:0> <EOL> now = time . time ( ) <EOL> nr_rows = table . get_nr_rows ( ) <EOL> properties_str = '<STR_LIT>' <EOL> if static : properties_str += '<STR_LIT>' <EOL> if update : properties_str += '<STR_LIT>' <EOL> if persist : properties_str += '<STR_LIT>' <EOL> logging . info ( "<STR_LIT>" , <EOL> properties_str , tablename , rowcount , identity , now - start_time ) <EOL> self . update_table_stats ( tablename , rowcount , identity = identity , <EOL> sum = not update and not static , now = now ) <EOL> return <NUM_LIT:1> </s>
