<s> from . _base import SQLBaseStore <EOL> from twisted . internet import defer <EOL> from canonicaljson import encode_canonical_json <EOL> import logging <EOL> import simplejson as json <EOL> import types <EOL> logger = logging . getLogger ( __name__ ) <EOL> class PusherStore ( SQLBaseStore ) : <EOL> def _decode_pushers_rows ( self , rows ) : <EOL> for r in rows : <EOL> dataJson = r [ '<STR_LIT:data>' ] <EOL> r [ '<STR_LIT:data>' ] = None <EOL> try : <EOL> if isinstance ( dataJson , types . BufferType ) : <EOL> dataJson = str ( dataJson ) . decode ( "<STR_LIT>" ) <EOL> r [ '<STR_LIT:data>' ] = json . loads ( dataJson ) <EOL> except Exception as e : <EOL> logger . warn ( <EOL> "<STR_LIT>" , <EOL> r [ '<STR_LIT:id>' ] , dataJson , e . message , <EOL> ) <EOL> pass <EOL> if isinstance ( r [ '<STR_LIT>' ] , types . BufferType ) : <EOL> r [ '<STR_LIT>' ] = str ( r [ '<STR_LIT>' ] ) . decode ( "<STR_LIT>" ) <EOL> return rows <EOL> @ defer . inlineCallbacks <EOL> def get_pushers_by_app_id_and_pushkey ( self , app_id , pushkey ) : <EOL> def r ( txn ) : <EOL> sql = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> txn . execute ( sql , ( app_id , pushkey , ) ) <EOL> rows = self . cursor_to_dict ( txn ) <EOL> return self . _decode_pushers_rows ( rows ) <EOL> rows = yield self . runInteraction ( <EOL> "<STR_LIT>" , r <EOL> ) <EOL> defer . returnValue ( rows ) <EOL> @ defer . inlineCallbacks <EOL> def get_all_pushers ( self ) : <EOL> def get_pushers ( txn ) : <EOL> txn . execute ( "<STR_LIT>" ) <EOL> rows = self . cursor_to_dict ( txn ) <EOL> return self . _decode_pushers_rows ( rows ) <EOL> rows = yield self . runInteraction ( "<STR_LIT>" , get_pushers ) <EOL> defer . returnValue ( rows ) <EOL> def get_pushers_stream_token ( self ) : <EOL> return self . _pushers_id_gen . get_max_token ( ) <EOL> def get_all_updated_pushers ( self , last_id , current_id , limit ) : <EOL> def get_all_updated_pushers_txn ( txn ) : <EOL> sql = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> txn . execute ( sql , ( last_id , current_id , limit ) ) <EOL> updated = txn . fetchall ( ) <EOL> sql = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> txn . execute ( sql , ( last_id , current_id , limit ) ) <EOL> deleted = txn . fetchall ( ) <EOL> return ( updated , deleted ) <EOL> return self . runInteraction ( <EOL> "<STR_LIT>" , get_all_updated_pushers_txn <EOL> ) <EOL> @ defer . inlineCallbacks <EOL> def add_pusher ( self , user_id , access_token , kind , app_id , <EOL> app_display_name , device_display_name , <EOL> pushkey , pushkey_ts , lang , data , profile_tag = "<STR_LIT>" ) : <EOL> with self . _pushers_id_gen . get_next ( ) as stream_id : <EOL> yield self . _simple_upsert ( <EOL> "<STR_LIT>" , <EOL> dict ( <EOL> app_id = app_id , <EOL> pushkey = pushkey , <EOL> user_name = user_id , <EOL> ) , <EOL> dict ( <EOL> access_token = access_token , <EOL> kind = kind , <EOL> app_display_name = app_display_name , <EOL> device_display_name = device_display_name , <EOL> ts = pushkey_ts , <EOL> lang = lang , <EOL> data = encode_canonical_json ( data ) , <EOL> profile_tag = profile_tag , <EOL> id = stream_id , <EOL> ) , <EOL> desc = "<STR_LIT>" , <EOL> ) <EOL> @ defer . inlineCallbacks <EOL> def delete_pusher_by_app_id_pushkey_user_id ( self , app_id , pushkey , user_id ) : <EOL> def delete_pusher_txn ( txn , stream_id ) : <EOL> self . _simple_delete_one_txn ( <EOL> txn , <EOL> "<STR_LIT>" , <EOL> { "<STR_LIT>" : app_id , "<STR_LIT>" : pushkey , "<STR_LIT>" : user_id } <EOL> ) <EOL> self . _simple_upsert_txn ( <EOL> txn , <EOL> "<STR_LIT>" , <EOL> { "<STR_LIT>" : app_id , "<STR_LIT>" : pushkey , "<STR_LIT>" : user_id } , <EOL> { "<STR_LIT>" : stream_id } , <EOL> ) <EOL> with self . _pushers_id_gen . get_next ( ) as stream_id : <EOL> yield self . runInteraction ( <EOL> "<STR_LIT>" , delete_pusher_txn , stream_id <EOL> ) <EOL> @ defer . inlineCallbacks <EOL> def update_pusher_last_token ( self , app_id , pushkey , user_id , last_token ) : <EOL> yield self . _simple_update_one ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : app_id , '<STR_LIT>' : pushkey , '<STR_LIT>' : user_id } , <EOL> { '<STR_LIT>' : last_token } , <EOL> desc = "<STR_LIT>" , <EOL> ) <EOL> @ defer . inlineCallbacks <EOL> def update_pusher_last_token_and_success ( self , app_id , pushkey , user_id , <EOL> last_token , last_success ) : <EOL> yield self . _simple_update_one ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : app_id , '<STR_LIT>' : pushkey , '<STR_LIT>' : user_id } , <EOL> { '<STR_LIT>' : last_token , '<STR_LIT>' : last_success } , <EOL> desc = "<STR_LIT>" , <EOL> ) <EOL> @ defer . inlineCallbacks <EOL> def update_pusher_failing_since ( self , app_id , pushkey , user_id , <EOL> failing_since ) : <EOL> yield self . _simple_update_one ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : app_id , '<STR_LIT>' : pushkey , '<STR_LIT>' : user_id } , <EOL> { '<STR_LIT>' : failing_since } , <EOL> desc = "<STR_LIT>" , <EOL> ) </s>
