<s> import logging <EOL> from synapse . api . errors import StoreError <EOL> from synapse . util . logcontext import LoggingContext , PreserveLoggingContext <EOL> from synapse . util . caches . dictionary_cache import DictionaryCache <EOL> from synapse . util . caches . descriptors import Cache <EOL> from synapse . util . caches import intern_dict <EOL> import synapse . metrics <EOL> from twisted . internet import defer <EOL> import sys <EOL> import time <EOL> import threading <EOL> import os <EOL> CACHE_SIZE_FACTOR = float ( os . environ . get ( "<STR_LIT>" , <NUM_LIT:0.1> ) ) <EOL> logger = logging . getLogger ( __name__ ) <EOL> sql_logger = logging . getLogger ( "<STR_LIT>" ) <EOL> transaction_logger = logging . getLogger ( "<STR_LIT>" ) <EOL> perf_logger = logging . getLogger ( "<STR_LIT>" ) <EOL> metrics = synapse . metrics . get_metrics_for ( "<STR_LIT>" ) <EOL> sql_scheduling_timer = metrics . register_distribution ( "<STR_LIT>" ) <EOL> sql_query_timer = metrics . register_distribution ( "<STR_LIT>" , labels = [ "<STR_LIT>" ] ) <EOL> sql_txn_timer = metrics . register_distribution ( "<STR_LIT>" , labels = [ "<STR_LIT>" ] ) <EOL> class LoggingTransaction ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = [ "<STR_LIT>" , "<STR_LIT:name>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> def __init__ ( self , txn , name , database_engine , after_callbacks ) : <EOL> object . __setattr__ ( self , "<STR_LIT>" , txn ) <EOL> object . __setattr__ ( self , "<STR_LIT:name>" , name ) <EOL> object . __setattr__ ( self , "<STR_LIT>" , database_engine ) <EOL> object . __setattr__ ( self , "<STR_LIT>" , after_callbacks ) <EOL> def call_after ( self , callback , * args ) : <EOL> """<STR_LIT>""" <EOL> self . after_callbacks . append ( ( callback , args ) ) <EOL> def __getattr__ ( self , name ) : <EOL> return getattr ( self . txn , name ) <EOL> def __setattr__ ( self , name , value ) : <EOL> setattr ( self . txn , name , value ) <EOL> def execute ( self , sql , * args ) : <EOL> self . _do_execute ( self . txn . execute , sql , * args ) <EOL> def executemany ( self , sql , * args ) : <EOL> self . _do_execute ( self . txn . executemany , sql , * args ) <EOL> def _do_execute ( self , func , sql , * args ) : <EOL> sql_logger . debug ( "<STR_LIT>" , self . name , sql ) <EOL> sql = self . database_engine . convert_param_style ( sql ) <EOL> if args : <EOL> try : <EOL> sql_logger . debug ( <EOL> "<STR_LIT>" , <EOL> self . name , args [ <NUM_LIT:0> ] <EOL> ) <EOL> except : <EOL> pass <EOL> start = time . time ( ) * <NUM_LIT:1000> <EOL> try : <EOL> return func ( <EOL> sql , * args <EOL> ) <EOL> except Exception as e : <EOL> logger . debug ( "<STR_LIT>" , self . name , e ) <EOL> raise <EOL> finally : <EOL> msecs = ( time . time ( ) * <NUM_LIT:1000> ) - start <EOL> sql_logger . debug ( "<STR_LIT>" , self . name , msecs ) <EOL> sql_query_timer . inc_by ( msecs , sql . split ( ) [ <NUM_LIT:0> ] ) <EOL> class PerformanceCounters ( object ) : <EOL> def __init__ ( self ) : <EOL> self . current_counters = { } <EOL> self . previous_counters = { } <EOL> def update ( self , key , start_time , end_time = None ) : <EOL> if end_time is None : <EOL> end_time = time . time ( ) * <NUM_LIT:1000> <EOL> duration = end_time - start_time <EOL> count , cum_time = self . current_counters . get ( key , ( <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> count += <NUM_LIT:1> <EOL> cum_time += duration <EOL> self . current_counters [ key ] = ( count , cum_time ) <EOL> return end_time <EOL> def interval ( self , interval_duration , limit = <NUM_LIT:3> ) : <EOL> counters = [ ] <EOL> for name , ( count , cum_time ) in self . current_counters . items ( ) : <EOL> prev_count , prev_time = self . previous_counters . get ( name , ( <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> counters . append ( ( <EOL> ( cum_time - prev_time ) / interval_duration , <EOL> count - prev_count , <EOL> name <EOL> ) ) <EOL> self . previous_counters = dict ( self . current_counters ) <EOL> counters . sort ( reverse = True ) <EOL> top_n_counters = "<STR_LIT:U+002CU+0020>" . join ( <EOL> "<STR_LIT>" % ( name , count , <NUM_LIT:100> * ratio ) <EOL> for ratio , count , name in counters [ : limit ] <EOL> ) <EOL> return top_n_counters <EOL> class SQLBaseStore ( object ) : <EOL> _TXN_ID = <NUM_LIT:0> <EOL> def __init__ ( self , hs ) : <EOL> self . hs = hs <EOL> self . _db_pool = hs . get_db_pool ( ) <EOL> self . _clock = hs . get_clock ( ) <EOL> self . _previous_txn_total_time = <NUM_LIT:0> <EOL> self . _current_txn_total_time = <NUM_LIT:0> <EOL> self . _previous_loop_ts = <NUM_LIT:0> <EOL> self . _txn_perf_counters = PerformanceCounters ( ) <EOL> self . _get_event_counters = PerformanceCounters ( ) <EOL> self . _get_event_cache = Cache ( "<STR_LIT>" , keylen = <NUM_LIT:3> , lru = True , <EOL> max_entries = hs . config . event_cache_size ) <EOL> self . _state_group_cache = DictionaryCache ( <EOL> "<STR_LIT>" , <NUM_LIT> * CACHE_SIZE_FACTOR <EOL> ) <EOL> self . _event_fetch_lock = threading . Condition ( ) <EOL> self . _event_fetch_list = [ ] <EOL> self . _event_fetch_ongoing = <NUM_LIT:0> <EOL> self . _pending_ds = [ ] <EOL> self . database_engine = hs . database_engine <EOL> def start_profiling ( self ) : <EOL> self . _previous_loop_ts = self . _clock . time_msec ( ) <EOL> def loop ( ) : <EOL> curr = self . _current_txn_total_time <EOL> prev = self . _previous_txn_total_time <EOL> self . _previous_txn_total_time = curr <EOL> time_now = self . _clock . time_msec ( ) <EOL> time_then = self . _previous_loop_ts <EOL> self . _previous_loop_ts = time_now <EOL> ratio = ( curr - prev ) / ( time_now - time_then ) <EOL> top_three_counters = self . _txn_perf_counters . interval ( <EOL> time_now - time_then , limit = <NUM_LIT:3> <EOL> ) <EOL> top_3_event_counters = self . _get_event_counters . interval ( <EOL> time_now - time_then , limit = <NUM_LIT:3> <EOL> ) <EOL> perf_logger . info ( <EOL> "<STR_LIT>" , <EOL> ratio * <NUM_LIT:100> , top_three_counters , top_3_event_counters <EOL> ) <EOL> self . _clock . looping_call ( loop , <NUM_LIT> ) <EOL> def _new_transaction ( self , conn , desc , after_callbacks , logging_context , <EOL> func , * args , ** kwargs ) : <EOL> start = time . time ( ) * <NUM_LIT:1000> <EOL> txn_id = self . _TXN_ID <EOL> self . _TXN_ID = ( self . _TXN_ID + <NUM_LIT:1> ) % ( sys . maxint - <NUM_LIT:1> ) <EOL> name = "<STR_LIT>" % ( desc , txn_id , ) <EOL> transaction_logger . debug ( "<STR_LIT>" , name ) <EOL> try : <EOL> i = <NUM_LIT:0> <EOL> N = <NUM_LIT:5> <EOL> while True : <EOL> try : <EOL> txn = conn . cursor ( ) <EOL> txn = LoggingTransaction ( <EOL> txn , name , self . database_engine , after_callbacks <EOL> ) <EOL> r = func ( txn , * args , ** kwargs ) <EOL> conn . commit ( ) <EOL> return r <EOL> except self . database_engine . module . OperationalError as e : <EOL> logger . warn ( <EOL> "<STR_LIT>" , <EOL> name , e , i , N <EOL> ) <EOL> if i < N : <EOL> i += <NUM_LIT:1> <EOL> try : <EOL> conn . rollback ( ) <EOL> except self . database_engine . module . Error as e1 : <EOL> logger . warn ( <EOL> "<STR_LIT>" , <EOL> name , e1 , <EOL> ) <EOL> continue <EOL> raise <EOL> except self . database_engine . module . DatabaseError as e : <EOL> if self . database_engine . is_deadlock ( e ) : <EOL> logger . warn ( "<STR_LIT>" , name , i , N ) <EOL> if i < N : <EOL> i += <NUM_LIT:1> <EOL> try : <EOL> conn . rollback ( ) <EOL> except self . database_engine . module . Error as e1 : <EOL> logger . warn ( <EOL> "<STR_LIT>" , <EOL> name , e1 , <EOL> ) <EOL> continue <EOL> raise <EOL> except Exception as e : <EOL> logger . debug ( "<STR_LIT>" , name , e ) <EOL> raise <EOL> finally : <EOL> end = time . time ( ) * <NUM_LIT:1000> <EOL> duration = end - start <EOL> if logging_context is not None : <EOL> logging_context . add_database_transaction ( duration ) <EOL> transaction_logger . debug ( "<STR_LIT>" , name , duration ) <EOL> self . _current_txn_total_time += duration <EOL> self . _txn_perf_counters . update ( desc , start , end ) <EOL> sql_txn_timer . inc_by ( duration , desc ) <EOL> @ defer . inlineCallbacks <EOL> def runInteraction ( self , desc , func , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> current_context = LoggingContext . current_context ( ) <EOL> start_time = time . time ( ) * <NUM_LIT:1000> <EOL> after_callbacks = [ ] <EOL> def inner_func ( conn , * args , ** kwargs ) : <EOL> with LoggingContext ( "<STR_LIT>" ) as context : <EOL> sql_scheduling_timer . inc_by ( time . time ( ) * <NUM_LIT:1000> - start_time ) <EOL> if self . database_engine . is_connection_closed ( conn ) : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> conn . reconnect ( ) <EOL> current_context . copy_to ( context ) <EOL> return self . _new_transaction ( <EOL> conn , desc , after_callbacks , current_context , <EOL> func , * args , ** kwargs <EOL> ) <EOL> with PreserveLoggingContext ( ) : <EOL> result = yield self . _db_pool . runWithConnection ( <EOL> inner_func , * args , ** kwargs <EOL> ) <EOL> for after_callback , after_args in after_callbacks : <EOL> after_callback ( * after_args ) <EOL> defer . returnValue ( result ) <EOL> @ defer . inlineCallbacks <EOL> def runWithConnection ( self , func , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> current_context = LoggingContext . current_context ( ) <EOL> start_time = time . time ( ) * <NUM_LIT:1000> <EOL> def inner_func ( conn , * args , ** kwargs ) : <EOL> with LoggingContext ( "<STR_LIT>" ) as context : <EOL> sql_scheduling_timer . inc_by ( time . time ( ) * <NUM_LIT:1000> - start_time ) <EOL> if self . database_engine . is_connection_closed ( conn ) : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> conn . reconnect ( ) <EOL> current_context . copy_to ( context ) <EOL> return func ( conn , * args , ** kwargs ) <EOL> with PreserveLoggingContext ( ) : <EOL> result = yield self . _db_pool . runWithConnection ( <EOL> inner_func , * args , ** kwargs <EOL> ) <EOL> defer . returnValue ( result ) <EOL> @ staticmethod <EOL> def cursor_to_dict ( cursor ) : <EOL> """<STR_LIT>""" <EOL> col_headers = list ( column [ <NUM_LIT:0> ] for column in cursor . description ) <EOL> results = list ( <EOL> intern_dict ( dict ( zip ( col_headers , row ) ) ) for row in cursor . fetchall ( ) <EOL> ) <EOL> return results <EOL> def _execute ( self , desc , decoder , query , * args ) : <EOL> """<STR_LIT>""" <EOL> def interaction ( txn ) : <EOL> txn . execute ( query , args ) <EOL> if decoder : <EOL> return decoder ( txn ) <EOL> else : <EOL> return txn . fetchall ( ) <EOL> return self . runInteraction ( desc , interaction ) <EOL> @ defer . inlineCallbacks <EOL> def _simple_insert ( self , table , values , or_ignore = False , <EOL> desc = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> yield self . runInteraction ( <EOL> desc , <EOL> self . _simple_insert_txn , table , values , <EOL> ) <EOL> except self . database_engine . module . IntegrityError : <EOL> if not or_ignore : <EOL> raise <EOL> @ staticmethod <EOL> def _simple_insert_txn ( txn , table , values ) : <EOL> keys , vals = zip ( * values . items ( ) ) <EOL> sql = "<STR_LIT>" % ( <EOL> table , <EOL> "<STR_LIT:U+002CU+0020>" . join ( k for k in keys ) , <EOL> "<STR_LIT:U+002CU+0020>" . join ( "<STR_LIT:?>" for _ in keys ) <EOL> ) <EOL> txn . execute ( sql , vals ) <EOL> @ staticmethod <EOL> def _simple_insert_many_txn ( txn , table , values ) : <EOL> if not values : <EOL> return <EOL> keys , vals = zip ( * [ <EOL> zip ( <EOL> * ( sorted ( i . items ( ) , key = lambda kv : kv [ <NUM_LIT:0> ] ) ) <EOL> ) <EOL> for i in values <EOL> if i <EOL> ] ) <EOL> for k in keys : <EOL> if k != keys [ <NUM_LIT:0> ] : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" <EOL> ) <EOL> sql = "<STR_LIT>" % ( <EOL> table , <EOL> "<STR_LIT:U+002CU+0020>" . join ( k for k in keys [ <NUM_LIT:0> ] ) , <EOL> "<STR_LIT:U+002CU+0020>" . join ( "<STR_LIT:?>" for _ in keys [ <NUM_LIT:0> ] ) <EOL> ) <EOL> txn . executemany ( sql , vals ) <EOL> def _simple_upsert ( self , table , keyvalues , values , <EOL> insertion_values = { } , desc = "<STR_LIT>" , lock = True ) : <EOL> """<STR_LIT>""" <EOL> return self . runInteraction ( <EOL> desc , <EOL> self . _simple_upsert_txn , table , keyvalues , values , insertion_values , <EOL> lock <EOL> ) <EOL> def _simple_upsert_txn ( self , txn , table , keyvalues , values , insertion_values = { } , <EOL> lock = True ) : <EOL> if lock : <EOL> self . database_engine . lock_table ( txn , table ) <EOL> sql = "<STR_LIT>" % ( <EOL> table , <EOL> "<STR_LIT:U+002CU+0020>" . join ( "<STR_LIT>" % ( k , ) for k in values ) , <EOL> "<STR_LIT>" . join ( "<STR_LIT>" % ( k , ) for k in keyvalues ) <EOL> ) <EOL> sqlargs = values . values ( ) + keyvalues . values ( ) <EOL> logger . debug ( <EOL> "<STR_LIT>" , <EOL> sql , sqlargs , <EOL> ) <EOL> txn . execute ( sql , sqlargs ) <EOL> if txn . rowcount == <NUM_LIT:0> : <EOL> allvalues = { } <EOL> allvalues . update ( keyvalues ) <EOL> allvalues . update ( values ) <EOL> allvalues . update ( insertion_values ) <EOL> sql = "<STR_LIT>" % ( <EOL> table , <EOL> "<STR_LIT:U+002CU+0020>" . join ( k for k in allvalues ) , <EOL> "<STR_LIT:U+002CU+0020>" . join ( "<STR_LIT:?>" for _ in allvalues ) <EOL> ) <EOL> logger . debug ( <EOL> "<STR_LIT>" , <EOL> sql , keyvalues . values ( ) , <EOL> ) <EOL> txn . execute ( sql , allvalues . values ( ) ) <EOL> def _simple_select_one ( self , table , keyvalues , retcols , <EOL> allow_none = False , desc = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> return self . runInteraction ( <EOL> desc , <EOL> self . _simple_select_one_txn , <EOL> table , keyvalues , retcols , allow_none , <EOL> ) <EOL> def _simple_select_one_onecol ( self , table , keyvalues , retcol , <EOL> allow_none = False , <EOL> desc = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> return self . runInteraction ( <EOL> desc , <EOL> self . _simple_select_one_onecol_txn , <EOL> table , keyvalues , retcol , allow_none = allow_none , <EOL> ) <EOL> @ classmethod <EOL> def _simple_select_one_onecol_txn ( cls , txn , table , keyvalues , retcol , <EOL> allow_none = False ) : <EOL> ret = cls . _simple_select_onecol_txn ( <EOL> txn , <EOL> table = table , <EOL> keyvalues = keyvalues , <EOL> retcol = retcol , <EOL> ) <EOL> if ret : <EOL> return ret [ <NUM_LIT:0> ] <EOL> else : <EOL> if allow_none : <EOL> return None <EOL> else : <EOL> raise StoreError ( <NUM_LIT> , "<STR_LIT>" ) <EOL> @ staticmethod <EOL> def _simple_select_onecol_txn ( txn , table , keyvalues , retcol ) : <EOL> sql = ( <EOL> "<STR_LIT>" <EOL> ) % { <EOL> "<STR_LIT>" : retcol , <EOL> "<STR_LIT>" : table , <EOL> "<STR_LIT>" : "<STR_LIT>" . join ( "<STR_LIT>" % k for k in keyvalues . keys ( ) ) , <EOL> } <EOL> txn . execute ( sql , keyvalues . values ( ) ) <EOL> return [ r [ <NUM_LIT:0> ] for r in txn . fetchall ( ) ] <EOL> def _simple_select_onecol ( self , table , keyvalues , retcol , <EOL> desc = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> return self . runInteraction ( <EOL> desc , <EOL> self . _simple_select_onecol_txn , <EOL> table , keyvalues , retcol <EOL> ) <EOL> def _simple_select_list ( self , table , keyvalues , retcols , <EOL> desc = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> return self . runInteraction ( <EOL> desc , <EOL> self . _simple_select_list_txn , <EOL> table , keyvalues , retcols <EOL> ) <EOL> @ classmethod <EOL> def _simple_select_list_txn ( cls , txn , table , keyvalues , retcols ) : <EOL> """<STR_LIT>""" <EOL> if keyvalues : <EOL> sql = "<STR_LIT>" % ( <EOL> "<STR_LIT:U+002CU+0020>" . join ( retcols ) , <EOL> table , <EOL> "<STR_LIT>" . join ( "<STR_LIT>" % ( k , ) for k in keyvalues ) <EOL> ) <EOL> txn . execute ( sql , keyvalues . values ( ) ) <EOL> else : <EOL> sql = "<STR_LIT>" % ( <EOL> "<STR_LIT:U+002CU+0020>" . join ( retcols ) , <EOL> table <EOL> ) <EOL> txn . execute ( sql ) <EOL> return cls . cursor_to_dict ( txn ) <EOL> @ defer . inlineCallbacks <EOL> def _simple_select_many_batch ( self , table , column , iterable , retcols , <EOL> keyvalues = { } , desc = "<STR_LIT>" , <EOL> batch_size = <NUM_LIT:100> ) : <EOL> """<STR_LIT>""" <EOL> results = [ ] <EOL> if not iterable : <EOL> defer . returnValue ( results ) <EOL> chunks = [ <EOL> iterable [ i : i + batch_size ] <EOL> for i in xrange ( <NUM_LIT:0> , len ( iterable ) , batch_size ) <EOL> ] <EOL> for chunk in chunks : <EOL> rows = yield self . runInteraction ( <EOL> desc , <EOL> self . _simple_select_many_txn , <EOL> table , column , chunk , keyvalues , retcols <EOL> ) <EOL> results . extend ( rows ) <EOL> defer . returnValue ( results ) <EOL> @ classmethod <EOL> def _simple_select_many_txn ( cls , txn , table , column , iterable , keyvalues , retcols ) : <EOL> """<STR_LIT>""" <EOL> if not iterable : <EOL> return [ ] <EOL> sql = "<STR_LIT>" % ( "<STR_LIT:U+002CU+0020>" . join ( retcols ) , table ) <EOL> clauses = [ ] <EOL> values = [ ] <EOL> clauses . append ( <EOL> "<STR_LIT>" % ( column , "<STR_LIT:U+002C>" . join ( "<STR_LIT:?>" for _ in iterable ) ) <EOL> ) <EOL> values . extend ( iterable ) <EOL> for key , value in keyvalues . items ( ) : <EOL> clauses . append ( "<STR_LIT>" % ( key , ) ) <EOL> values . append ( value ) <EOL> if clauses : <EOL> sql = "<STR_LIT>" % ( <EOL> sql , <EOL> "<STR_LIT>" . join ( clauses ) , <EOL> ) <EOL> txn . execute ( sql , values ) <EOL> return cls . cursor_to_dict ( txn ) <EOL> def _simple_update_one ( self , table , keyvalues , updatevalues , <EOL> desc = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> return self . runInteraction ( <EOL> desc , <EOL> self . _simple_update_one_txn , <EOL> table , keyvalues , updatevalues , <EOL> ) <EOL> @ staticmethod <EOL> def _simple_update_one_txn ( txn , table , keyvalues , updatevalues ) : <EOL> update_sql = "<STR_LIT>" % ( <EOL> table , <EOL> "<STR_LIT:U+002CU+0020>" . join ( "<STR_LIT>" % ( k , ) for k in updatevalues ) , <EOL> "<STR_LIT>" . join ( "<STR_LIT>" % ( k , ) for k in keyvalues ) <EOL> ) <EOL> txn . execute ( <EOL> update_sql , <EOL> updatevalues . values ( ) + keyvalues . values ( ) <EOL> ) <EOL> if txn . rowcount == <NUM_LIT:0> : <EOL> raise StoreError ( <NUM_LIT> , "<STR_LIT>" ) <EOL> if txn . rowcount > <NUM_LIT:1> : <EOL> raise StoreError ( <NUM_LIT> , "<STR_LIT>" ) <EOL> @ staticmethod <EOL> def _simple_select_one_txn ( txn , table , keyvalues , retcols , <EOL> allow_none = False ) : <EOL> select_sql = "<STR_LIT>" % ( <EOL> "<STR_LIT:U+002CU+0020>" . join ( retcols ) , <EOL> table , <EOL> "<STR_LIT>" . join ( "<STR_LIT>" % ( k , ) for k in keyvalues ) <EOL> ) <EOL> txn . execute ( select_sql , keyvalues . values ( ) ) <EOL> row = txn . fetchone ( ) <EOL> if not row : <EOL> if allow_none : <EOL> return None <EOL> raise StoreError ( <NUM_LIT> , "<STR_LIT>" ) <EOL> if txn . rowcount > <NUM_LIT:1> : <EOL> raise StoreError ( <NUM_LIT> , "<STR_LIT>" ) <EOL> return dict ( zip ( retcols , row ) ) <EOL> def _simple_delete_one ( self , table , keyvalues , desc = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> return self . runInteraction ( <EOL> desc , self . _simple_delete_one_txn , table , keyvalues <EOL> ) <EOL> @ staticmethod <EOL> def _simple_delete_one_txn ( txn , table , keyvalues ) : <EOL> """<STR_LIT>""" <EOL> sql = "<STR_LIT>" % ( <EOL> table , <EOL> "<STR_LIT>" . join ( "<STR_LIT>" % ( k , ) for k in keyvalues ) <EOL> ) <EOL> txn . execute ( sql , keyvalues . values ( ) ) <EOL> if txn . rowcount == <NUM_LIT:0> : <EOL> raise StoreError ( <NUM_LIT> , "<STR_LIT>" ) <EOL> if txn . rowcount > <NUM_LIT:1> : <EOL> raise StoreError ( <NUM_LIT> , "<STR_LIT>" ) <EOL> @ staticmethod <EOL> def _simple_delete_txn ( txn , table , keyvalues ) : <EOL> sql = "<STR_LIT>" % ( <EOL> table , <EOL> "<STR_LIT>" . join ( "<STR_LIT>" % ( k , ) for k in keyvalues ) <EOL> ) <EOL> return txn . execute ( sql , keyvalues . values ( ) ) <EOL> def get_next_stream_id ( self ) : <EOL> with self . _next_stream_id_lock : <EOL> i = self . _next_stream_id <EOL> self . _next_stream_id += <NUM_LIT:1> <EOL> return i <EOL> class _RollbackButIsFineException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass </s>
