<s> """<STR_LIT>""" <EOL> import logging <EOL> import sys <EOL> from google . appengine . api import datastore <EOL> from google . appengine . api import datastore_errors <EOL> from google . appengine . api import memcache <EOL> from google . appengine . api import namespace_manager <EOL> from google . appengine . datastore import datastore_rpc <EOL> from google . appengine . datastore import entity_pb <EOL> from google . net . proto import ProtocolBuffer <EOL> from . import key as key_module <EOL> from . import model <EOL> from . import tasklets <EOL> from . import eventloop <EOL> from . import utils <EOL> logging_debug = utils . logging_debug <EOL> _LOCK_TIME = <NUM_LIT:32> <EOL> _LOCKED = <NUM_LIT:0> <EOL> class ContextOptions ( datastore_rpc . TransactionOptions ) : <EOL> """<STR_LIT>""" <EOL> @ datastore_rpc . ConfigOption <EOL> def use_cache ( value ) : <EOL> if not isinstance ( value , bool ) : <EOL> raise datastore_errors . BadArgumentError ( <EOL> '<STR_LIT>' % ( value , ) ) <EOL> return value <EOL> @ datastore_rpc . ConfigOption <EOL> def use_memcache ( value ) : <EOL> if not isinstance ( value , bool ) : <EOL> raise datastore_errors . BadArgumentError ( <EOL> '<STR_LIT>' % ( value , ) ) <EOL> return value <EOL> @ datastore_rpc . ConfigOption <EOL> def use_datastore ( value ) : <EOL> if not isinstance ( value , bool ) : <EOL> raise datastore_errors . BadArgumentError ( <EOL> '<STR_LIT>' % ( value , ) ) <EOL> return value <EOL> @ datastore_rpc . ConfigOption <EOL> def memcache_timeout ( value ) : <EOL> if not isinstance ( value , ( int , long ) ) : <EOL> raise datastore_errors . BadArgumentError ( <EOL> '<STR_LIT>' % ( value , ) ) <EOL> return value <EOL> @ datastore_rpc . ConfigOption <EOL> def max_memcache_items ( value ) : <EOL> if not isinstance ( value , ( int , long ) ) : <EOL> raise datastore_errors . BadArgumentError ( <EOL> '<STR_LIT>' % ( value , ) ) <EOL> return value <EOL> _OPTION_TRANSLATIONS = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> def _make_ctx_options ( ctx_options ) : <EOL> """<STR_LIT>""" <EOL> if not ctx_options : <EOL> return None <EOL> for key in list ( ctx_options ) : <EOL> translation = _OPTION_TRANSLATIONS . get ( key ) <EOL> if translation : <EOL> if translation in ctx_options : <EOL> raise ValueError ( '<STR_LIT>' % <EOL> ( key , translation ) ) <EOL> ctx_options [ translation ] = ctx_options . pop ( key ) <EOL> return ContextOptions ( ** ctx_options ) <EOL> class AutoBatcher ( object ) : <EOL> def __init__ ( self , todo_tasklet , limit ) : <EOL> self . _todo_tasklet = todo_tasklet <EOL> self . _limit = limit <EOL> self . _queues = { } <EOL> self . _running = [ ] <EOL> self . _cache = { } <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , self . _todo_tasklet . __name__ ) <EOL> def run_queue ( self , options , todo ) : <EOL> logging_debug ( '<STR_LIT>' , <EOL> self . _todo_tasklet . __name__ , len ( todo ) ) <EOL> fut = self . _todo_tasklet ( todo , options ) <EOL> self . _running . append ( fut ) <EOL> fut . add_callback ( self . _finished_callback , fut ) <EOL> def _on_idle ( self ) : <EOL> if not self . action ( ) : <EOL> return None <EOL> return True <EOL> def add ( self , arg , options = None ) : <EOL> fut = tasklets . Future ( '<STR_LIT>' % ( self , arg , options ) ) <EOL> todo = self . _queues . get ( options ) <EOL> if todo is None : <EOL> logging_debug ( '<STR_LIT>' , <EOL> self . _todo_tasklet . __name__ , options ) <EOL> if not self . _queues : <EOL> eventloop . add_idle ( self . _on_idle ) <EOL> todo = self . _queues [ options ] = [ ] <EOL> todo . append ( ( fut , arg ) ) <EOL> if len ( todo ) >= self . _limit : <EOL> del self . _queues [ options ] <EOL> self . run_queue ( options , todo ) <EOL> return fut <EOL> def add_once ( self , arg , options = None ) : <EOL> cache_key = ( arg , options ) <EOL> fut = self . _cache . get ( cache_key ) <EOL> if fut is None : <EOL> fut = self . add ( arg , options ) <EOL> self . _cache [ cache_key ] = fut <EOL> fut . add_immediate_callback ( self . _cache . __delitem__ , cache_key ) <EOL> return fut <EOL> def action ( self ) : <EOL> queues = self . _queues <EOL> if not queues : <EOL> return False <EOL> options , todo = queues . popitem ( ) <EOL> self . run_queue ( options , todo ) <EOL> return True <EOL> def _finished_callback ( self , fut ) : <EOL> self . _running . remove ( fut ) <EOL> fut . check_success ( ) <EOL> @ tasklets . tasklet <EOL> def flush ( self ) : <EOL> while self . _running or self . action ( ) : <EOL> if self . _running : <EOL> yield self . _running <EOL> class Context ( object ) : <EOL> def __init__ ( self , conn = None , auto_batcher_class = AutoBatcher , config = None ) : <EOL> if conn is None : <EOL> conn = model . make_connection ( config ) <EOL> self . _conn = conn <EOL> self . _auto_batcher_class = auto_batcher_class <EOL> max_get = ( datastore_rpc . Configuration . max_get_keys ( config , conn . config ) or <EOL> datastore_rpc . Connection . MAX_GET_KEYS ) <EOL> max_put = ( datastore_rpc . Configuration . max_put_entities ( config , <EOL> conn . config ) or <EOL> datastore_rpc . Connection . MAX_PUT_ENTITIES ) <EOL> max_delete = ( datastore_rpc . Configuration . max_delete_keys ( config , <EOL> conn . config ) or <EOL> datastore_rpc . Connection . MAX_DELETE_KEYS ) <EOL> self . _get_batcher = auto_batcher_class ( self . _get_tasklet , max_get ) <EOL> self . _put_batcher = auto_batcher_class ( self . _put_tasklet , max_put ) <EOL> self . _delete_batcher = auto_batcher_class ( self . _delete_tasklet , max_delete ) <EOL> max_memcache = ( ContextOptions . max_memcache_items ( config , conn . config ) or <EOL> datastore_rpc . Connection . MAX_GET_KEYS ) <EOL> self . _memcache_get_batcher = auto_batcher_class ( self . _memcache_get_tasklet , <EOL> max_memcache ) <EOL> self . _memcache_set_batcher = auto_batcher_class ( self . _memcache_set_tasklet , <EOL> max_memcache ) <EOL> self . _memcache_del_batcher = auto_batcher_class ( self . _memcache_del_tasklet , <EOL> max_memcache ) <EOL> self . _memcache_off_batcher = auto_batcher_class ( self . _memcache_off_tasklet , <EOL> max_memcache ) <EOL> self . _batchers = [ self . _get_batcher , <EOL> self . _put_batcher , <EOL> self . _delete_batcher , <EOL> self . _memcache_get_batcher , <EOL> self . _memcache_set_batcher , <EOL> self . _memcache_del_batcher , <EOL> self . _memcache_off_batcher , <EOL> ] <EOL> self . _cache = { } <EOL> self . _memcache = memcache . Client ( ) <EOL> _memcache_prefix = '<STR_LIT>' <EOL> @ tasklets . tasklet <EOL> def flush ( self ) : <EOL> more = True <EOL> while more : <EOL> yield [ batcher . flush ( ) for batcher in self . _batchers ] <EOL> more = False <EOL> for batcher in self . _batchers : <EOL> if batcher . _running or batcher . _queues : <EOL> more = True <EOL> break <EOL> @ tasklets . tasklet <EOL> def _get_tasklet ( self , todo , options ) : <EOL> if not todo : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> datastore_keys = [ ] <EOL> for unused_fut , key in todo : <EOL> datastore_keys . append ( key ) <EOL> entities = yield self . _conn . async_get ( options , datastore_keys ) <EOL> for ent , ( fut , unused_key ) in zip ( entities , todo ) : <EOL> fut . set_result ( ent ) <EOL> @ tasklets . tasklet <EOL> def _put_tasklet ( self , todo , options ) : <EOL> if not todo : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> datastore_entities = [ ] <EOL> for unused_fut , ent in todo : <EOL> datastore_entities . append ( ent ) <EOL> keys = yield self . _conn . async_put ( options , datastore_entities ) <EOL> for key , ( fut , ent ) in zip ( keys , todo ) : <EOL> if key != ent . _key : <EOL> if ent . _has_complete_key ( ) : <EOL> raise datastore_errors . BadKeyError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( key , ent . _key ) ) <EOL> ent . _key = key <EOL> fut . set_result ( key ) <EOL> @ tasklets . tasklet <EOL> def _delete_tasklet ( self , todo , options ) : <EOL> if not todo : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> futures = [ ] <EOL> datastore_keys = [ ] <EOL> for fut , key in todo : <EOL> futures . append ( fut ) <EOL> datastore_keys . append ( key ) <EOL> yield self . _conn . async_delete ( options , datastore_keys ) <EOL> for fut in futures : <EOL> fut . set_result ( None ) <EOL> @ staticmethod <EOL> def default_cache_policy ( key ) : <EOL> """<STR_LIT>""" <EOL> flag = None <EOL> if key is not None : <EOL> modelclass = model . Model . _kind_map . get ( key . kind ( ) ) <EOL> if modelclass is not None : <EOL> policy = getattr ( modelclass , '<STR_LIT>' , None ) <EOL> if policy is not None : <EOL> if isinstance ( policy , bool ) : <EOL> flag = policy <EOL> else : <EOL> flag = policy ( key ) <EOL> return flag <EOL> _cache_policy = default_cache_policy <EOL> def get_cache_policy ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _cache_policy <EOL> def set_cache_policy ( self , func ) : <EOL> """<STR_LIT>""" <EOL> if func is None : <EOL> func = self . default_cache_policy <EOL> elif isinstance ( func , bool ) : <EOL> func = lambda unused_key , flag = func : flag <EOL> self . _cache_policy = func <EOL> def _use_cache ( self , key , options = None ) : <EOL> """<STR_LIT>""" <EOL> flag = ContextOptions . use_cache ( options ) <EOL> if flag is None : <EOL> flag = self . _cache_policy ( key ) <EOL> if flag is None : <EOL> flag = ContextOptions . use_cache ( self . _conn . config ) <EOL> if flag is None : <EOL> flag = True <EOL> return flag <EOL> @ staticmethod <EOL> def default_memcache_policy ( key ) : <EOL> """<STR_LIT>""" <EOL> flag = None <EOL> if key is not None : <EOL> modelclass = model . Model . _kind_map . get ( key . kind ( ) ) <EOL> if modelclass is not None : <EOL> policy = getattr ( modelclass , '<STR_LIT>' , None ) <EOL> if policy is not None : <EOL> if isinstance ( policy , bool ) : <EOL> flag = policy <EOL> else : <EOL> flag = policy ( key ) <EOL> return flag <EOL> _memcache_policy = default_memcache_policy <EOL> def get_memcache_policy ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _memcache_policy <EOL> def set_memcache_policy ( self , func ) : <EOL> """<STR_LIT>""" <EOL> if func is None : <EOL> func = self . default_memcache_policy <EOL> elif isinstance ( func , bool ) : <EOL> func = lambda unused_key , flag = func : flag <EOL> self . _memcache_policy = func <EOL> def _use_memcache ( self , key , options = None ) : <EOL> """<STR_LIT>""" <EOL> flag = ContextOptions . use_memcache ( options ) <EOL> if flag is None : <EOL> flag = self . _memcache_policy ( key ) <EOL> if flag is None : <EOL> flag = ContextOptions . use_memcache ( self . _conn . config ) <EOL> if flag is None : <EOL> flag = True <EOL> return flag <EOL> @ staticmethod <EOL> def default_datastore_policy ( key ) : <EOL> """<STR_LIT>""" <EOL> flag = None <EOL> if key is not None : <EOL> modelclass = model . Model . _kind_map . get ( key . kind ( ) ) <EOL> if modelclass is not None : <EOL> policy = getattr ( modelclass , '<STR_LIT>' , None ) <EOL> if policy is not None : <EOL> if isinstance ( policy , bool ) : <EOL> flag = policy <EOL> else : <EOL> flag = policy ( key ) <EOL> return flag <EOL> _datastore_policy = default_datastore_policy <EOL> def get_datastore_policy ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _datastore_policy <EOL> def set_datastore_policy ( self , func ) : <EOL> """<STR_LIT>""" <EOL> if func is None : <EOL> func = self . default_datastore_policy <EOL> elif isinstance ( func , bool ) : <EOL> func = lambda unused_key , flag = func : flag <EOL> self . _datastore_policy = func <EOL> def _use_datastore ( self , key , options = None ) : <EOL> """<STR_LIT>""" <EOL> flag = ContextOptions . use_datastore ( options ) <EOL> if flag is None : <EOL> flag = self . _datastore_policy ( key ) <EOL> if flag is None : <EOL> flag = ContextOptions . use_datastore ( self . _conn . config ) <EOL> if flag is None : <EOL> flag = True <EOL> return flag <EOL> @ staticmethod <EOL> def default_memcache_timeout_policy ( key ) : <EOL> """<STR_LIT>""" <EOL> timeout = None <EOL> if key is not None and isinstance ( key , model . Key ) : <EOL> modelclass = model . Model . _kind_map . get ( key . kind ( ) ) <EOL> if modelclass is not None : <EOL> policy = getattr ( modelclass , '<STR_LIT>' , None ) <EOL> if policy is not None : <EOL> if isinstance ( policy , ( int , long ) ) : <EOL> timeout = policy <EOL> else : <EOL> timeout = policy ( key ) <EOL> return timeout <EOL> _memcache_timeout_policy = default_memcache_timeout_policy <EOL> def set_memcache_timeout_policy ( self , func ) : <EOL> """<STR_LIT>""" <EOL> if func is None : <EOL> func = self . default_memcache_timeout_policy <EOL> elif isinstance ( func , ( int , long ) ) : <EOL> func = lambda unused_key , flag = func : flag <EOL> self . _memcache_timeout_policy = func <EOL> def get_memcache_timeout_policy ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _memcache_timeout_policy <EOL> def _get_memcache_timeout ( self , key , options = None ) : <EOL> """<STR_LIT>""" <EOL> timeout = ContextOptions . memcache_timeout ( options ) <EOL> if timeout is None : <EOL> timeout = self . _memcache_timeout_policy ( key ) <EOL> if timeout is None : <EOL> timeout = ContextOptions . memcache_timeout ( self . _conn . config ) <EOL> if timeout is None : <EOL> timeout = <NUM_LIT:0> <EOL> return timeout <EOL> @ tasklets . tasklet <EOL> def get ( self , key , ** ctx_options ) : <EOL> """<STR_LIT>""" <EOL> options = _make_ctx_options ( ctx_options ) <EOL> use_cache = self . _use_cache ( key , options ) <EOL> if use_cache : <EOL> if key in self . _cache : <EOL> entity = self . _cache [ key ] <EOL> if entity is None or entity . _key == key : <EOL> raise tasklets . Return ( entity ) <EOL> use_datastore = self . _use_datastore ( key , options ) <EOL> use_memcache = self . _use_memcache ( key , options ) <EOL> using_tconn = isinstance ( self . _conn , datastore_rpc . TransactionalConnection ) <EOL> in_transaction = ( use_datastore and using_tconn ) <EOL> ns = key . namespace ( ) <EOL> if use_memcache and not in_transaction : <EOL> mkey = self . _memcache_prefix + key . urlsafe ( ) <EOL> mvalue = yield self . memcache_get ( mkey , for_cas = use_datastore , <EOL> namespace = ns , use_cache = True ) <EOL> if mvalue not in ( _LOCKED , None ) : <EOL> cls = model . Model . _kind_map . get ( key . kind ( ) ) <EOL> if cls is None : <EOL> raise TypeError ( '<STR_LIT>' % key . kind ( ) ) <EOL> pb = entity_pb . EntityProto ( ) <EOL> try : <EOL> pb . MergePartialFromString ( mvalue ) <EOL> except ProtocolBuffer . ProtocolBufferDecodeError : <EOL> logging . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( mkey , ns ) ) <EOL> mvalue = None <EOL> else : <EOL> entity = cls . _from_pb ( pb ) <EOL> entity . _key = key <EOL> raise tasklets . Return ( entity ) <EOL> if mvalue is None and use_datastore : <EOL> yield self . memcache_set ( mkey , _LOCKED , time = _LOCK_TIME , namespace = ns , <EOL> use_cache = True ) <EOL> yield self . memcache_gets ( mkey , namespace = ns , use_cache = True ) <EOL> if not use_datastore : <EOL> raise tasklets . Return ( None ) <EOL> if use_cache : <EOL> entity = yield self . _get_batcher . add_once ( key , options ) <EOL> else : <EOL> entity = yield self . _get_batcher . add ( key , options ) <EOL> if entity is not None : <EOL> if not in_transaction and use_memcache and mvalue != _LOCKED : <EOL> pbs = entity . _to_pb ( set_key = False ) . SerializePartialToString ( ) <EOL> timeout = self . _get_memcache_timeout ( key , options ) <EOL> self . memcache_cas ( mkey , pbs , time = timeout , namespace = ns ) <EOL> if use_cache : <EOL> self . _cache [ key ] = entity <EOL> raise tasklets . Return ( entity ) <EOL> @ tasklets . tasklet <EOL> def put ( self , entity , ** ctx_options ) : <EOL> options = _make_ctx_options ( ctx_options ) <EOL> key = entity . _key <EOL> if key is None : <EOL> key = model . Key ( entity . __class__ , None ) <EOL> use_datastore = self . _use_datastore ( key , options ) <EOL> if entity . _has_complete_key ( ) : <EOL> if self . _use_memcache ( key , options ) : <EOL> mkey = self . _memcache_prefix + key . urlsafe ( ) <EOL> ns = key . namespace ( ) <EOL> if use_datastore : <EOL> yield self . memcache_set ( mkey , _LOCKED , time = _LOCK_TIME , <EOL> namespace = ns , use_cache = True ) <EOL> else : <EOL> pbs = entity . _to_pb ( set_key = False ) . SerializePartialToString ( ) <EOL> timeout = self . _get_memcache_timeout ( key , options ) <EOL> yield self . memcache_set ( mkey , pbs , time = timeout , namespace = ns ) <EOL> if use_datastore : <EOL> key = yield self . _put_batcher . add ( entity , options ) <EOL> if self . _use_memcache ( key , options ) : <EOL> mkey = self . _memcache_prefix + key . urlsafe ( ) <EOL> ns = key . namespace ( ) <EOL> yield self . memcache_delete ( mkey , namespace = ns ) <EOL> if key is not None : <EOL> if entity . _key != key : <EOL> logging . info ( '<STR_LIT>' , entity . _key , key ) <EOL> entity . _key = key <EOL> if self . _use_cache ( key , options ) : <EOL> self . _cache [ key ] = entity <EOL> raise tasklets . Return ( key ) <EOL> @ tasklets . tasklet <EOL> def delete ( self , key , ** ctx_options ) : <EOL> options = _make_ctx_options ( ctx_options ) <EOL> if self . _use_memcache ( key , options ) : <EOL> mkey = self . _memcache_prefix + key . urlsafe ( ) <EOL> ns = key . namespace ( ) <EOL> yield self . memcache_set ( mkey , _LOCKED , time = _LOCK_TIME , namespace = ns , <EOL> use_cache = True ) <EOL> if self . _use_datastore ( key , options ) : <EOL> yield self . _delete_batcher . add ( key , options ) <EOL> if self . _use_cache ( key , options ) : <EOL> self . _cache [ key ] = None <EOL> @ tasklets . tasklet <EOL> def allocate_ids ( self , key , size = None , max = None , ** ctx_options ) : <EOL> options = _make_ctx_options ( ctx_options ) <EOL> lo_hi = yield self . _conn . async_allocate_ids ( options , key , size , max ) <EOL> raise tasklets . Return ( lo_hi ) <EOL> @ datastore_rpc . _positional ( <NUM_LIT:3> ) <EOL> def map_query ( self , query , callback , options = None , merge_future = None ) : <EOL> mfut = merge_future <EOL> if mfut is None : <EOL> mfut = tasklets . MultiFuture ( '<STR_LIT>' ) <EOL> @ tasklets . tasklet <EOL> def helper ( ) : <EOL> try : <EOL> inq = tasklets . SerialQueueFuture ( ) <EOL> query . run_to_queue ( inq , self . _conn , options ) <EOL> is_ancestor_query = query . ancestor is not None <EOL> while True : <EOL> try : <EOL> batch , i , ent = yield inq . getq ( ) <EOL> except EOFError : <EOL> break <EOL> if isinstance ( ent , model . Key ) : <EOL> pass <EOL> else : <EOL> key = ent . _key <EOL> if key in self . _cache : <EOL> hit = self . _cache [ key ] <EOL> if hit is not None and hit . key != key : <EOL> del self . _cache [ key ] <EOL> if key in self . _cache : <EOL> if self . _cache [ key ] is None : <EOL> logging . info ( '<STR_LIT>' , key ) <EOL> continue <EOL> if ent != self . _cache [ key ] : <EOL> logging . info ( '<STR_LIT>' , key ) <EOL> ent = self . _cache [ key ] <EOL> else : <EOL> if is_ancestor_query and self . _use_cache ( key , options ) : <EOL> self . _cache [ key ] = ent <EOL> if callback is None : <EOL> val = ent <EOL> else : <EOL> if options is not None and options . produce_cursors : <EOL> val = callback ( batch , i , ent ) <EOL> else : <EOL> val = callback ( ent ) <EOL> mfut . putq ( val ) <EOL> except Exception , err : <EOL> _ , _ , tb = sys . exc_info ( ) <EOL> mfut . set_exception ( err , tb ) <EOL> raise <EOL> else : <EOL> mfut . complete ( ) <EOL> helper ( ) <EOL> return mfut <EOL> @ datastore_rpc . _positional ( <NUM_LIT:2> ) <EOL> def iter_query ( self , query , callback = None , options = None ) : <EOL> return self . map_query ( query , callback = callback , options = options , <EOL> merge_future = tasklets . SerialQueueFuture ( ) ) <EOL> @ tasklets . tasklet <EOL> def transaction ( self , callback , ** ctx_options ) : <EOL> options = _make_ctx_options ( ctx_options ) <EOL> app = ContextOptions . app ( options ) or key_module . _DefaultAppId ( ) <EOL> retries = ContextOptions . retries ( options ) <EOL> if retries is None : <EOL> retries = <NUM_LIT:3> <EOL> yield self . flush ( ) <EOL> for _ in xrange ( <NUM_LIT:1> + max ( <NUM_LIT:0> , retries ) ) : <EOL> transaction = yield self . _conn . async_begin_transaction ( options , app ) <EOL> tconn = datastore_rpc . TransactionalConnection ( <EOL> adapter = self . _conn . adapter , <EOL> config = self . _conn . config , <EOL> transaction = transaction ) <EOL> old_ds_conn = datastore . _GetConnection ( ) <EOL> tctx = self . __class__ ( conn = tconn , <EOL> auto_batcher_class = self . _auto_batcher_class ) <EOL> try : <EOL> tctx . set_memcache_policy ( self . get_memcache_policy ( ) ) <EOL> tctx . set_memcache_timeout_policy ( self . get_memcache_timeout_policy ( ) ) <EOL> tasklets . set_context ( tctx ) <EOL> datastore . _SetConnection ( tconn ) <EOL> try : <EOL> try : <EOL> result = callback ( ) <EOL> if isinstance ( result , tasklets . Future ) : <EOL> result = yield result <EOL> finally : <EOL> yield tctx . flush ( ) <EOL> except Exception : <EOL> t , e , tb = sys . exc_info ( ) <EOL> yield tconn . async_rollback ( options ) <EOL> if issubclass ( t , datastore_errors . Rollback ) : <EOL> return <EOL> else : <EOL> raise t , e , tb <EOL> else : <EOL> ok = yield tconn . async_commit ( options ) <EOL> if ok : <EOL> self . _cache . update ( tctx . _cache ) <EOL> yield self . _clear_memcache ( tctx . _cache ) <EOL> raise tasklets . Return ( result ) <EOL> finally : <EOL> datastore . _SetConnection ( old_ds_conn ) <EOL> raise datastore_errors . TransactionFailedError ( <EOL> '<STR_LIT>' ) <EOL> def in_transaction ( self ) : <EOL> """<STR_LIT>""" <EOL> return isinstance ( self . _conn , datastore_rpc . TransactionalConnection ) <EOL> def clear_cache ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _cache . clear ( ) <EOL> self . _get_queue . clear ( ) <EOL> @ tasklets . tasklet <EOL> def _clear_memcache ( self , keys ) : <EOL> keys = set ( key for key in keys if self . _use_memcache ( key ) ) <EOL> futures = [ ] <EOL> for key in keys : <EOL> mkey = self . _memcache_prefix + key . urlsafe ( ) <EOL> ns = key . namespace ( ) <EOL> fut = self . memcache_set ( mkey , _LOCKED , time = _LOCK_TIME , namespace = ns , <EOL> use_cache = True ) <EOL> futures . append ( fut ) <EOL> yield futures <EOL> @ tasklets . tasklet <EOL> def get_or_insert ( * args , ** kwds ) : <EOL> self , model_class , name = args <EOL> our_kwds = { } <EOL> for kwd in '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' : <EOL> alt_kwd = '<STR_LIT:_>' + kwd <EOL> if alt_kwd in kwds : <EOL> our_kwds [ kwd ] = kwds . pop ( alt_kwd ) <EOL> elif ( kwd in kwds and <EOL> not isinstance ( getattr ( model_class , kwd , None ) , model . Property ) ) : <EOL> our_kwds [ kwd ] = kwds . pop ( kwd ) <EOL> app = our_kwds . get ( '<STR_LIT>' ) <EOL> namespace = our_kwds . get ( '<STR_LIT>' ) <EOL> parent = our_kwds . get ( '<STR_LIT>' ) <EOL> context_options = our_kwds . get ( '<STR_LIT>' ) <EOL> if not isinstance ( name , basestring ) : <EOL> raise TypeError ( '<STR_LIT>' % name ) <EOL> elif not name : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> key = model . Key ( model_class , name , <EOL> app = app , namespace = namespace , parent = parent ) <EOL> ent = yield self . get ( key ) <EOL> if ent is None : <EOL> @ tasklets . tasklet <EOL> def txn ( ) : <EOL> ent = yield key . get_async ( options = context_options ) <EOL> if ent is None : <EOL> ent = model_class ( ** kwds ) <EOL> ent . _key = key <EOL> yield ent . put_async ( options = context_options ) <EOL> raise tasklets . Return ( ent ) <EOL> ent = yield self . transaction ( txn ) <EOL> raise tasklets . Return ( ent ) <EOL> @ tasklets . tasklet <EOL> def _memcache_get_tasklet ( self , todo , options ) : <EOL> if not todo : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> for_cas , namespace = options <EOL> keys = set ( ) <EOL> for unused_fut , key in todo : <EOL> keys . add ( key ) <EOL> results = yield self . _memcache . get_multi_async ( keys , for_cas = for_cas , <EOL> namespace = namespace ) <EOL> for fut , key in todo : <EOL> fut . set_result ( results . get ( key ) ) <EOL> @ tasklets . tasklet <EOL> def _memcache_set_tasklet ( self , todo , options ) : <EOL> if not todo : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> opname , time , namespace = options <EOL> methodname = opname + '<STR_LIT>' <EOL> method = getattr ( self . _memcache , methodname ) <EOL> mapping = { } <EOL> for unused_fut , ( key , value ) in todo : <EOL> mapping [ key ] = value <EOL> results = yield method ( mapping , time = time , namespace = namespace ) <EOL> for fut , ( key , unused_value ) in todo : <EOL> if results is None : <EOL> status = memcache . MemcacheSetResponse . ERROR <EOL> else : <EOL> status = results . get ( key ) <EOL> fut . set_result ( status == memcache . MemcacheSetResponse . STORED ) <EOL> @ tasklets . tasklet <EOL> def _memcache_del_tasklet ( self , todo , options ) : <EOL> if not todo : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> seconds , namespace = options <EOL> keys = set ( ) <EOL> for unused_fut , key in todo : <EOL> keys . add ( key ) <EOL> statuses = yield self . _memcache . delete_multi_async ( keys , seconds = seconds , <EOL> namespace = namespace ) <EOL> status_key_mapping = { } <EOL> if statuses : <EOL> for key , status in zip ( keys , statuses ) : <EOL> status_key_mapping [ key ] = status <EOL> for fut , key in todo : <EOL> status = status_key_mapping . get ( key , memcache . DELETE_NETWORK_FAILURE ) <EOL> fut . set_result ( status ) <EOL> @ tasklets . tasklet <EOL> def _memcache_off_tasklet ( self , todo , options ) : <EOL> if not todo : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> initial_value , namespace = options <EOL> mapping = { } <EOL> for unused_fut , ( key , delta ) in todo : <EOL> mapping [ key ] = delta <EOL> results = yield self . _memcache . offset_multi_async ( mapping , <EOL> initial_value = initial_value , namespace = namespace ) <EOL> for fut , ( key , unused_delta ) in todo : <EOL> result = results . get ( key ) <EOL> if isinstance ( result , basestring ) : <EOL> result = int ( result ) <EOL> fut . set_result ( result ) <EOL> def memcache_get ( self , key , for_cas = False , namespace = None , use_cache = False ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( key , str ) : <EOL> raise TypeError ( '<STR_LIT>' % key ) <EOL> if not isinstance ( for_cas , bool ) : <EOL> raise ValueError ( '<STR_LIT>' % for_cas ) <EOL> if namespace is None : <EOL> namespace = namespace_manager . get_namespace ( ) <EOL> options = ( for_cas , namespace ) <EOL> batcher = self . _memcache_get_batcher <EOL> if use_cache : <EOL> return batcher . add_once ( key , options ) <EOL> else : <EOL> return batcher . add ( key , options ) <EOL> def memcache_gets ( self , key , namespace = None , use_cache = False ) : <EOL> return self . memcache_get ( key , for_cas = True , namespace = namespace , <EOL> use_cache = use_cache ) <EOL> def memcache_set ( self , key , value , time = <NUM_LIT:0> , namespace = None , use_cache = False ) : <EOL> if not isinstance ( key , str ) : <EOL> raise TypeError ( '<STR_LIT>' % key ) <EOL> if not isinstance ( time , ( int , long ) ) : <EOL> raise ValueError ( '<STR_LIT>' % time ) <EOL> if namespace is None : <EOL> namespace = namespace_manager . get_namespace ( ) <EOL> options = ( '<STR_LIT>' , time , namespace ) <EOL> batcher = self . _memcache_set_batcher <EOL> if use_cache : <EOL> return batcher . add_once ( ( key , value ) , options ) <EOL> else : <EOL> return batcher . add ( ( key , value ) , options ) <EOL> def memcache_add ( self , key , value , time = <NUM_LIT:0> , namespace = None ) : <EOL> if not isinstance ( key , str ) : <EOL> raise TypeError ( '<STR_LIT>' % key ) <EOL> if not isinstance ( time , ( int , long ) ) : <EOL> raise ValueError ( '<STR_LIT>' % time ) <EOL> if namespace is None : <EOL> namespace = namespace_manager . get_namespace ( ) <EOL> return self . _memcache_set_batcher . add ( ( key , value ) , <EOL> ( '<STR_LIT>' , time , namespace ) ) <EOL> def memcache_replace ( self , key , value , time = <NUM_LIT:0> , namespace = None ) : <EOL> if not isinstance ( key , str ) : <EOL> raise TypeError ( '<STR_LIT>' % key ) <EOL> if not isinstance ( time , ( int , long ) ) : <EOL> raise ValueError ( '<STR_LIT>' % time ) <EOL> if namespace is None : <EOL> namespace = namespace_manager . get_namespace ( ) <EOL> return self . _memcache_set_batcher . add ( ( key , value ) , <EOL> ( '<STR_LIT:replace>' , time , namespace ) ) <EOL> def memcache_cas ( self , key , value , time = <NUM_LIT:0> , namespace = None ) : <EOL> if not isinstance ( key , str ) : <EOL> raise TypeError ( '<STR_LIT>' % key ) <EOL> if not isinstance ( time , ( int , long ) ) : <EOL> raise ValueError ( '<STR_LIT>' % time ) <EOL> if namespace is None : <EOL> namespace = namespace_manager . get_namespace ( ) <EOL> return self . _memcache_set_batcher . add ( ( key , value ) , <EOL> ( '<STR_LIT>' , time , namespace ) ) <EOL> def memcache_delete ( self , key , seconds = <NUM_LIT:0> , namespace = None ) : <EOL> if not isinstance ( key , str ) : <EOL> raise TypeError ( '<STR_LIT>' % key ) <EOL> if not isinstance ( seconds , ( int , long ) ) : <EOL> raise ValueError ( '<STR_LIT>' % seconds ) <EOL> if namespace is None : <EOL> namespace = namespace_manager . get_namespace ( ) <EOL> return self . _memcache_del_batcher . add ( key , ( seconds , namespace ) ) <EOL> def memcache_incr ( self , key , delta = <NUM_LIT:1> , initial_value = None , namespace = None ) : <EOL> if not isinstance ( key , str ) : <EOL> raise TypeError ( '<STR_LIT>' % key ) <EOL> if not isinstance ( delta , ( int , long ) ) : <EOL> raise ValueError ( '<STR_LIT>' % delta ) <EOL> if initial_value is not None and not isinstance ( initial_value , ( int , long ) ) : <EOL> raise ValueError ( '<STR_LIT>' % <EOL> initial_value ) <EOL> if namespace is None : <EOL> namespace = namespace_manager . get_namespace ( ) <EOL> return self . _memcache_off_batcher . add ( ( key , delta ) , <EOL> ( initial_value , namespace ) ) <EOL> def memcache_decr ( self , key , delta = <NUM_LIT:1> , initial_value = None , namespace = None ) : <EOL> if not isinstance ( key , str ) : <EOL> raise TypeError ( '<STR_LIT>' % key ) <EOL> if not isinstance ( delta , ( int , long ) ) : <EOL> raise ValueError ( '<STR_LIT>' % delta ) <EOL> if initial_value is not None and not isinstance ( initial_value , ( int , long ) ) : <EOL> raise ValueError ( '<STR_LIT>' % <EOL> initial_value ) <EOL> if namespace is None : <EOL> namespace = namespace_manager . get_namespace ( ) <EOL> return self . _memcache_off_batcher . add ( ( key , - delta ) , <EOL> ( initial_value , namespace ) ) <EOL> def toplevel ( func ) : <EOL> """<STR_LIT>""" <EOL> @ utils . wrapping ( func ) <EOL> def add_context_wrapper ( * args , ** kwds ) : <EOL> __ndb_debug__ = utils . func_info ( func ) <EOL> tasklets . _state . clear_all_pending ( ) <EOL> ctx = tasklets . make_default_context ( ) <EOL> try : <EOL> tasklets . set_context ( ctx ) <EOL> return tasklets . synctasklet ( func ) ( * args , ** kwds ) <EOL> finally : <EOL> tasklets . set_context ( None ) <EOL> ctx . flush ( ) . check_success ( ) <EOL> eventloop . run ( ) <EOL> return add_context_wrapper </s>
