<s> """<STR_LIT>""" <EOL> import apsw <EOL> import json <EOL> import time <EOL> import traceback <EOL> from bayeslite import IBayesDBTracer <EOL> from bayeslite . loggers import BqlLogger , CallHomeStatusLogger <EOL> from bayeslite . schema import bayesdb_schema_required <EOL> from bayeslite . util import cursor_value <EOL> from bayeslite import __version__ <EOL> _error_previous_session_msg = '<STR_LIT>' <EOL> class SessionOrchestrator ( object ) : <EOL> def __init__ ( self , bdb , meta_logger = None , session_logger = None ) : <EOL> bayesdb_schema_required ( bdb , <NUM_LIT:7> , '<STR_LIT>' ) <EOL> self . bdb = bdb <EOL> if meta_logger is None : <EOL> meta_logger = BqlLogger ( ) <EOL> self . _logger = meta_logger <EOL> if session_logger is None : <EOL> session_logger = CallHomeStatusLogger ( ) <EOL> self . _session_logger = session_logger <EOL> self . _qid_to_entry_id = { } <EOL> self . _sql_tracer = _SessionTracer ( "<STR_LIT>" , self ) <EOL> self . _bql_tracer = _SessionTracer ( "<STR_LIT>" , self ) <EOL> self . start_saving_sessions ( ) <EOL> self . _suggested_send = False <EOL> self . _start_new_session ( ) <EOL> def _sql ( self , query , bindings = None ) : <EOL> if bindings == None : <EOL> bindings = ( ) <EOL> return self . bdb . _sqlite3 . cursor ( ) . execute ( query , bindings ) <EOL> def _add_entry ( self , qid , type , query , bindings ) : <EOL> '''<STR_LIT>''' <EOL> if not self . _suggested_send : <EOL> self . _check_error_entries ( self . session_id ) <EOL> data = query + json . dumps ( bindings ) <EOL> self . _sql ( '''<STR_LIT>''' , ( self . session_id , type , data , time . time ( ) ) ) <EOL> entry_id = cursor_value ( self . _sql ( '<STR_LIT>' ) ) <EOL> self . _qid_to_entry_id [ qid ] = entry_id <EOL> def _mark_entry_completed ( self , qid ) : <EOL> entry_id = self . _qid_to_entry_id [ qid ] <EOL> self . _sql ( '''<STR_LIT>''' , ( time . time ( ) , entry_id ) ) <EOL> def _mark_entry_error ( self , qid ) : <EOL> entry_id = self . _qid_to_entry_id [ qid ] <EOL> self . _sql ( '''<STR_LIT>''' , ( traceback . format_exc ( ) , time . time ( ) , entry_id ) ) <EOL> def _start_new_session ( self ) : <EOL> self . _sql ( '<STR_LIT>' , <EOL> ( __version__ , ) ) <EOL> self . session_id = cursor_value ( self . _sql ( '<STR_LIT>' ) ) <EOL> self . _check_error_entries ( self . session_id - <NUM_LIT:1> ) <EOL> def _check_error_entries ( self , session_id ) : <EOL> '''<STR_LIT>''' <EOL> error_entries = cursor_value ( self . _sql ( '''<STR_LIT>''' , ( session_id , ) ) ) <EOL> if ( error_entries > <NUM_LIT:0> and not self . _suggested_send ) : <EOL> self . _logger . warn ( _error_previous_session_msg ) <EOL> self . _suggested_send = True <EOL> return error_entries <EOL> def clear_all_sessions ( self ) : <EOL> self . _sql ( '<STR_LIT>' ) <EOL> self . _sql ( '<STR_LIT>' ) <EOL> self . _sql ( '''<STR_LIT>''' ) <EOL> self . _start_new_session ( ) <EOL> def list_sessions ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _sql ( '<STR_LIT>' ) <EOL> def current_session_id ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . session_id <EOL> def dump_session_as_json ( self , session_id ) : <EOL> """<STR_LIT>""" <EOL> if session_id > self . session_id or session_id < <NUM_LIT:1> : <EOL> raise ValueError ( '<STR_LIT>' % session_id ) <EOL> version = cursor_value ( self . _sql ( '''<STR_LIT>''' , ( session_id , ) ) ) <EOL> cursor = self . _sql ( '''<STR_LIT>''' , ( session_id , ) ) <EOL> fields = [ ] <EOL> try : <EOL> fields = [ d [ <NUM_LIT:0> ] for d in cursor . description ] <EOL> except apsw . ExecutionCompleteError : <EOL> pass <EOL> entries = cursor . fetchall ( ) <EOL> session = { <EOL> '<STR_LIT>' : entries , <EOL> '<STR_LIT>' : fields , <EOL> '<STR_LIT:version>' : version , <EOL> } <EOL> return json . dumps ( session , sort_keys = True ) <EOL> def dump_current_session_as_json ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . dump_session_as_json ( self . session_id ) <EOL> def send_session_data ( self ) : <EOL> """<STR_LIT>""" <EOL> for id in range ( <NUM_LIT:1> , self . session_id + <NUM_LIT:1> ) : <EOL> self . _logger . info ( '<STR_LIT>' , id ) <EOL> json_string = self . dump_session_as_json ( id ) <EOL> self . _logger . info ( json_string ) <EOL> self . _session_logger . _send ( json_string ) <EOL> def start_saving_sessions ( self ) : <EOL> self . bdb . trace ( self . _bql_tracer ) <EOL> self . bdb . sql_trace ( self . _sql_tracer ) <EOL> def stop_saving_sessions ( self ) : <EOL> self . bdb . untrace ( self . _bql_tracer ) <EOL> self . bdb . sql_untrace ( self . _sql_tracer ) <EOL> class _SessionTracer ( IBayesDBTracer ) : <EOL> def __init__ ( self , type , orchestrator ) : <EOL> self . _type = type <EOL> self . _orchestrator = orchestrator <EOL> def start ( self , qid , query , bindings ) : <EOL> self . _orchestrator . _add_entry ( qid , self . _type , query , bindings ) <EOL> def finished ( self , qid ) : <EOL> self . _orchestrator . _mark_entry_completed ( qid ) <EOL> def error ( self , qid , e ) : <EOL> self . _orchestrator . _mark_entry_error ( qid ) </s>
