<s> """<STR_LIT>""" <EOL> __all__ = [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> import cgi <EOL> import datetime <EOL> import urllib <EOL> import zlib <EOL> import google <EOL> from google . appengine . _internal . graphy . backends import google_chart_api <EOL> import simplejson <EOL> from google . appengine . api import memcache <EOL> from google . appengine . api import taskqueue <EOL> from google . appengine . datastore import datastore_rpc <EOL> from google . appengine . ext import db <EOL> from google . appengine . ext . mapreduce import context <EOL> from google . appengine . ext . mapreduce import hooks <EOL> from google . appengine . ext . mapreduce import json_util <EOL> from google . appengine . ext . mapreduce import util <EOL> _MAP_REDUCE_KINDS = ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> class _HugeTaskPayload ( db . Model ) : <EOL> """<STR_LIT>""" <EOL> payload = db . BlobProperty ( ) <EOL> @ classmethod <EOL> def kind ( cls ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" <EOL> class HugeTask ( object ) : <EOL> """<STR_LIT>""" <EOL> PAYLOAD_PARAM = "<STR_LIT>" <EOL> PAYLOAD_KEY_PARAM = "<STR_LIT>" <EOL> MAX_TASK_PAYLOAD = taskqueue . MAX_PUSH_TASK_SIZE_BYTES - <NUM_LIT> <EOL> MAX_DB_PAYLOAD = datastore_rpc . BaseConnection . MAX_RPC_BYTES <EOL> PAYLOAD_VERSION_HEADER = "<STR_LIT>" <EOL> PAYLOAD_VERSION = "<STR_LIT:1>" <EOL> def __init__ ( self , <EOL> url , <EOL> params , <EOL> name = None , <EOL> eta = None , <EOL> countdown = None , <EOL> parent = None , <EOL> headers = None ) : <EOL> """<STR_LIT>""" <EOL> self . url = url <EOL> self . name = name <EOL> self . eta = eta <EOL> self . countdown = countdown <EOL> self . _headers = { <EOL> "<STR_LIT:Content-Type>" : "<STR_LIT>" , <EOL> self . PAYLOAD_VERSION_HEADER : self . PAYLOAD_VERSION <EOL> } <EOL> if headers : <EOL> self . _headers . update ( headers ) <EOL> payload_str = urllib . urlencode ( params ) <EOL> compressed_payload = "<STR_LIT>" <EOL> if len ( payload_str ) > self . MAX_TASK_PAYLOAD : <EOL> compressed_payload = zlib . compress ( payload_str ) <EOL> if not compressed_payload : <EOL> self . _payload = payload_str <EOL> elif len ( compressed_payload ) < self . MAX_TASK_PAYLOAD : <EOL> self . _payload = self . PAYLOAD_PARAM + compressed_payload <EOL> elif len ( compressed_payload ) > self . MAX_DB_PAYLOAD : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % <EOL> ( self . name , len ( compressed_payload ) ) ) <EOL> else : <EOL> if not parent : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> payload_entity = _HugeTaskPayload ( payload = compressed_payload , <EOL> parent = parent ) <EOL> payload_key = payload_entity . put ( ) <EOL> self . _payload = self . PAYLOAD_KEY_PARAM + str ( payload_key ) <EOL> def add ( self , queue_name , transactional = False ) : <EOL> """<STR_LIT>""" <EOL> task = self . to_task ( ) <EOL> task . add ( queue_name , transactional ) <EOL> def to_task ( self ) : <EOL> """<STR_LIT>""" <EOL> return taskqueue . Task ( <EOL> url = self . url , <EOL> payload = self . _payload , <EOL> name = self . name , <EOL> eta = self . eta , <EOL> countdown = self . countdown , <EOL> headers = self . _headers ) <EOL> @ classmethod <EOL> def decode_payload ( cls , request ) : <EOL> """<STR_LIT>""" <EOL> if request . headers . get ( cls . PAYLOAD_VERSION_HEADER ) != cls . PAYLOAD_VERSION : <EOL> raise DeprecationWarning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return cls . _decode_payload ( request . body ) <EOL> @ classmethod <EOL> def _decode_payload ( cls , body ) : <EOL> compressed_payload_str = None <EOL> if body . startswith ( cls . PAYLOAD_KEY_PARAM ) : <EOL> payload_key = body [ len ( cls . PAYLOAD_KEY_PARAM ) : ] <EOL> payload_entity = _HugeTaskPayload . get ( payload_key ) <EOL> compressed_payload_str = payload_entity . payload <EOL> elif body . startswith ( cls . PAYLOAD_PARAM ) : <EOL> compressed_payload_str = body [ len ( cls . PAYLOAD_PARAM ) : ] <EOL> if compressed_payload_str : <EOL> payload_str = zlib . decompress ( compressed_payload_str ) <EOL> else : <EOL> payload_str = body <EOL> result = { } <EOL> for ( name , value ) in cgi . parse_qs ( payload_str ) . items ( ) : <EOL> if len ( value ) == <NUM_LIT:1> : <EOL> result [ name ] = value [ <NUM_LIT:0> ] <EOL> else : <EOL> result [ name ] = value <EOL> return result <EOL> class CountersMap ( json_util . JsonMixin ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , initial_map = None ) : <EOL> """<STR_LIT>""" <EOL> if initial_map : <EOL> self . counters = initial_map <EOL> else : <EOL> self . counters = { } <EOL> def __repr__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % self . counters <EOL> def get ( self , counter_name , default = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> return self . counters . get ( counter_name , default ) <EOL> def increment ( self , counter_name , delta ) : <EOL> """<STR_LIT>""" <EOL> current_value = self . counters . get ( counter_name , <NUM_LIT:0> ) <EOL> new_value = current_value + delta <EOL> self . counters [ counter_name ] = new_value <EOL> return new_value <EOL> def add_map ( self , counters_map ) : <EOL> """<STR_LIT>""" <EOL> for counter_name in counters_map . counters : <EOL> self . increment ( counter_name , counters_map . counters [ counter_name ] ) <EOL> def sub_map ( self , counters_map ) : <EOL> """<STR_LIT>""" <EOL> for counter_name in counters_map . counters : <EOL> self . increment ( counter_name , - counters_map . counters [ counter_name ] ) <EOL> def clear ( self ) : <EOL> """<STR_LIT>""" <EOL> self . counters = { } <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> return { "<STR_LIT>" : self . counters } <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> """<STR_LIT>""" <EOL> counters_map = cls ( ) <EOL> counters_map . counters = json [ "<STR_LIT>" ] <EOL> return counters_map <EOL> def to_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . counters <EOL> class MapperSpec ( json_util . JsonMixin ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> handler_spec , <EOL> input_reader_spec , <EOL> params , <EOL> shard_count , <EOL> output_writer_spec = None ) : <EOL> """<STR_LIT>""" <EOL> self . handler_spec = handler_spec <EOL> self . input_reader_spec = input_reader_spec <EOL> self . output_writer_spec = output_writer_spec <EOL> self . shard_count = int ( shard_count ) <EOL> self . params = params <EOL> def get_handler ( self ) : <EOL> """<STR_LIT>""" <EOL> return util . handler_for_name ( self . handler_spec ) <EOL> handler = property ( get_handler ) <EOL> def input_reader_class ( self ) : <EOL> """<STR_LIT>""" <EOL> return util . for_name ( self . input_reader_spec ) <EOL> def output_writer_class ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . output_writer_spec and util . for_name ( self . output_writer_spec ) <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> result = { <EOL> "<STR_LIT>" : self . handler_spec , <EOL> "<STR_LIT>" : self . input_reader_spec , <EOL> "<STR_LIT>" : self . params , <EOL> "<STR_LIT>" : self . shard_count <EOL> } <EOL> if self . output_writer_spec : <EOL> result [ "<STR_LIT>" ] = self . output_writer_spec <EOL> return result <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % ( <EOL> self . handler_spec , self . input_reader_spec , self . params , <EOL> self . shard_count ) <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> """<STR_LIT>""" <EOL> return cls ( json [ "<STR_LIT>" ] , <EOL> json [ "<STR_LIT>" ] , <EOL> json [ "<STR_LIT>" ] , <EOL> json [ "<STR_LIT>" ] , <EOL> json . get ( "<STR_LIT>" ) <EOL> ) <EOL> def __eq__ ( self , other ) : <EOL> if not isinstance ( other , self . __class__ ) : <EOL> return False <EOL> return self . to_json ( ) == other . to_json ( ) <EOL> class MapreduceSpec ( json_util . JsonMixin ) : <EOL> """<STR_LIT>""" <EOL> PARAM_DONE_CALLBACK = "<STR_LIT>" <EOL> PARAM_DONE_CALLBACK_QUEUE = "<STR_LIT>" <EOL> PARAM_DONE_CALLBACK_TARGET = "<STR_LIT>" <EOL> PARAM_MAPREDUCE_TARGET = "<STR_LIT>" <EOL> def __init__ ( self , <EOL> name , <EOL> mapreduce_id , <EOL> mapper_spec , <EOL> params = { } , <EOL> hooks_class_name = None ) : <EOL> """<STR_LIT>""" <EOL> self . name = name <EOL> self . mapreduce_id = mapreduce_id <EOL> self . mapper = MapperSpec . from_json ( mapper_spec ) <EOL> self . params = params <EOL> self . hooks_class_name = hooks_class_name <EOL> self . __hooks = None <EOL> self . get_hooks ( ) <EOL> def get_hooks ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . __hooks is None and self . hooks_class_name is not None : <EOL> hooks_class = util . for_name ( self . hooks_class_name ) <EOL> if not isinstance ( hooks_class , type ) : <EOL> raise ValueError ( "<STR_LIT>" % <EOL> type ( hooks_class ) . __name__ ) <EOL> if not issubclass ( hooks_class , hooks . Hooks ) : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> self . __hooks = hooks_class ( self ) <EOL> return self . __hooks <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> mapper_spec = self . mapper . to_json ( ) <EOL> return { <EOL> "<STR_LIT:name>" : self . name , <EOL> "<STR_LIT>" : self . mapreduce_id , <EOL> "<STR_LIT>" : mapper_spec , <EOL> "<STR_LIT>" : self . params , <EOL> "<STR_LIT>" : self . hooks_class_name , <EOL> } <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> """<STR_LIT>""" <EOL> mapreduce_spec = cls ( json [ "<STR_LIT:name>" ] , <EOL> json [ "<STR_LIT>" ] , <EOL> json [ "<STR_LIT>" ] , <EOL> json . get ( "<STR_LIT>" ) , <EOL> json . get ( "<STR_LIT>" ) ) <EOL> return mapreduce_spec <EOL> def __str__ ( self ) : <EOL> return str ( self . to_json ( ) ) <EOL> def __eq__ ( self , other ) : <EOL> if not isinstance ( other , self . __class__ ) : <EOL> return False <EOL> return self . to_json ( ) == other . to_json ( ) <EOL> @ classmethod <EOL> def _get_mapreduce_spec ( cls , mr_id ) : <EOL> """<STR_LIT>""" <EOL> key = '<STR_LIT>' % mr_id <EOL> spec_json = memcache . get ( key ) <EOL> if spec_json : <EOL> return cls . from_json ( spec_json ) <EOL> state = MapreduceState . get_by_job_id ( mr_id ) <EOL> spec = state . mapreduce_spec <EOL> spec_json = spec . to_json ( ) <EOL> memcache . set ( key , spec_json ) <EOL> return spec <EOL> class MapreduceState ( db . Model ) : <EOL> """<STR_LIT>""" <EOL> RESULT_SUCCESS = "<STR_LIT:success>" <EOL> RESULT_FAILED = "<STR_LIT>" <EOL> RESULT_ABORTED = "<STR_LIT>" <EOL> _RESULTS = frozenset ( [ RESULT_SUCCESS , RESULT_FAILED , RESULT_ABORTED ] ) <EOL> mapreduce_spec = json_util . JsonProperty ( MapreduceSpec , indexed = False ) <EOL> active = db . BooleanProperty ( default = True , indexed = False ) <EOL> last_poll_time = db . DateTimeProperty ( required = True ) <EOL> counters_map = json_util . JsonProperty ( <EOL> CountersMap , default = CountersMap ( ) , indexed = False ) <EOL> app_id = db . StringProperty ( required = False , indexed = True ) <EOL> writer_state = json_util . JsonProperty ( dict , indexed = False ) <EOL> active_shards = db . IntegerProperty ( default = <NUM_LIT:0> , indexed = False ) <EOL> failed_shards = db . IntegerProperty ( default = <NUM_LIT:0> , indexed = False ) <EOL> aborted_shards = db . IntegerProperty ( default = <NUM_LIT:0> , indexed = False ) <EOL> result_status = db . StringProperty ( required = False , choices = _RESULTS ) <EOL> chart_url = db . TextProperty ( default = "<STR_LIT>" ) <EOL> chart_width = db . IntegerProperty ( default = <NUM_LIT> , indexed = False ) <EOL> sparkline_url = db . TextProperty ( default = "<STR_LIT>" ) <EOL> start_time = db . DateTimeProperty ( auto_now_add = True ) <EOL> @ classmethod <EOL> def kind ( cls ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" <EOL> @ classmethod <EOL> def get_key_by_job_id ( cls , mapreduce_id ) : <EOL> """<STR_LIT>""" <EOL> return db . Key . from_path ( cls . kind ( ) , str ( mapreduce_id ) ) <EOL> @ classmethod <EOL> def get_by_job_id ( cls , mapreduce_id ) : <EOL> """<STR_LIT>""" <EOL> return db . get ( cls . get_key_by_job_id ( mapreduce_id ) ) <EOL> def set_processed_counts ( self , shards_processed ) : <EOL> """<STR_LIT>""" <EOL> chart = google_chart_api . BarChart ( shards_processed ) <EOL> shard_count = len ( shards_processed ) <EOL> if shards_processed : <EOL> stride_length = max ( <NUM_LIT:1> , shard_count / <NUM_LIT:16> ) <EOL> chart . bottom . labels = [ ] <EOL> for x in xrange ( shard_count ) : <EOL> if ( x % stride_length == <NUM_LIT:0> or <EOL> x == shard_count - <NUM_LIT:1> ) : <EOL> chart . bottom . labels . append ( x ) <EOL> else : <EOL> chart . bottom . labels . append ( "<STR_LIT>" ) <EOL> chart . left . labels = [ "<STR_LIT:0>" , str ( max ( shards_processed ) ) ] <EOL> chart . left . min = <NUM_LIT:0> <EOL> self . chart_width = min ( <NUM_LIT> , max ( <NUM_LIT> , shard_count * <NUM_LIT:20> ) ) <EOL> self . chart_url = chart . display . Url ( self . chart_width , <NUM_LIT:200> ) <EOL> def get_processed ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . counters_map . get ( context . COUNTER_MAPPER_CALLS ) <EOL> processed = property ( get_processed ) <EOL> @ staticmethod <EOL> def create_new ( mapreduce_id = None , <EOL> gettime = datetime . datetime . now ) : <EOL> """<STR_LIT>""" <EOL> if not mapreduce_id : <EOL> mapreduce_id = MapreduceState . new_mapreduce_id ( ) <EOL> state = MapreduceState ( key_name = mapreduce_id , <EOL> last_poll_time = gettime ( ) ) <EOL> state . set_processed_counts ( [ ] ) <EOL> return state <EOL> @ staticmethod <EOL> def new_mapreduce_id ( ) : <EOL> """<STR_LIT>""" <EOL> return util . _get_descending_key ( ) <EOL> def __eq__ ( self , other ) : <EOL> if not isinstance ( other , self . __class__ ) : <EOL> return False <EOL> return self . properties ( ) == other . properties ( ) <EOL> class TransientShardState ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> base_path , <EOL> mapreduce_spec , <EOL> shard_id , <EOL> slice_id , <EOL> input_reader , <EOL> initial_input_reader , <EOL> output_writer = None , <EOL> retries = <NUM_LIT:0> , <EOL> handler = None ) : <EOL> """<STR_LIT>""" <EOL> self . base_path = base_path <EOL> self . mapreduce_spec = mapreduce_spec <EOL> self . shard_id = shard_id <EOL> self . slice_id = slice_id <EOL> self . input_reader = input_reader <EOL> self . initial_input_reader = initial_input_reader <EOL> self . output_writer = output_writer <EOL> self . retries = retries <EOL> self . handler = handler <EOL> self . _input_reader_json = self . input_reader . to_json ( ) <EOL> def reset_for_retry ( self , output_writer ) : <EOL> """<STR_LIT>""" <EOL> self . input_reader = self . initial_input_reader <EOL> self . slice_id = <NUM_LIT:0> <EOL> self . retries += <NUM_LIT:1> <EOL> self . output_writer = output_writer <EOL> self . handler = self . mapreduce_spec . mapper . handler <EOL> def advance_for_next_slice ( self , recovery_slice = False ) : <EOL> """<STR_LIT>""" <EOL> if recovery_slice : <EOL> self . slice_id += <NUM_LIT:2> <EOL> self . input_reader = self . input_reader . from_json ( self . _input_reader_json ) <EOL> else : <EOL> self . slice_id += <NUM_LIT:1> <EOL> def to_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> result = { "<STR_LIT>" : self . mapreduce_spec . to_json_str ( ) , <EOL> "<STR_LIT>" : self . shard_id , <EOL> "<STR_LIT>" : str ( self . slice_id ) , <EOL> "<STR_LIT>" : self . input_reader . to_json_str ( ) , <EOL> "<STR_LIT>" : <EOL> self . initial_input_reader . to_json_str ( ) , <EOL> "<STR_LIT>" : str ( self . retries ) } <EOL> if self . output_writer : <EOL> result [ "<STR_LIT>" ] = self . output_writer . to_json_str ( ) <EOL> serialized_handler = util . try_serialize_handler ( self . handler ) <EOL> if serialized_handler : <EOL> result [ "<STR_LIT>" ] = serialized_handler <EOL> return result <EOL> @ classmethod <EOL> def from_request ( cls , request ) : <EOL> """<STR_LIT>""" <EOL> mapreduce_spec = MapreduceSpec . from_json_str ( request . get ( "<STR_LIT>" ) ) <EOL> mapper_spec = mapreduce_spec . mapper <EOL> input_reader_spec_dict = simplejson . loads ( request . get ( "<STR_LIT>" ) , <EOL> cls = json_util . JsonDecoder ) <EOL> input_reader = mapper_spec . input_reader_class ( ) . from_json ( <EOL> input_reader_spec_dict ) <EOL> initial_input_reader_spec_dict = simplejson . loads ( <EOL> request . get ( "<STR_LIT>" ) , cls = json_util . JsonDecoder ) <EOL> initial_input_reader = mapper_spec . input_reader_class ( ) . from_json ( <EOL> initial_input_reader_spec_dict ) <EOL> output_writer = None <EOL> if mapper_spec . output_writer_class ( ) : <EOL> output_writer = mapper_spec . output_writer_class ( ) . from_json ( <EOL> simplejson . loads ( request . get ( "<STR_LIT>" , "<STR_LIT:{}>" ) , <EOL> cls = json_util . JsonDecoder ) ) <EOL> assert isinstance ( output_writer , mapper_spec . output_writer_class ( ) ) , ( <EOL> "<STR_LIT>" % ( <EOL> mapper_spec . output_writer_class ( ) , <EOL> output_writer . __class__ ) ) <EOL> handler = util . try_deserialize_handler ( request . get ( "<STR_LIT>" ) ) <EOL> if not handler : <EOL> handler = mapreduce_spec . mapper . handler <EOL> return cls ( mapreduce_spec . params [ "<STR_LIT>" ] , <EOL> mapreduce_spec , <EOL> str ( request . get ( "<STR_LIT>" ) ) , <EOL> int ( request . get ( "<STR_LIT>" ) ) , <EOL> input_reader , <EOL> initial_input_reader , <EOL> output_writer = output_writer , <EOL> retries = int ( request . get ( "<STR_LIT>" ) ) , <EOL> handler = handler ) <EOL> class ShardState ( db . Model ) : <EOL> """<STR_LIT>""" <EOL> RESULT_SUCCESS = "<STR_LIT:success>" <EOL> RESULT_FAILED = "<STR_LIT>" <EOL> RESULT_ABORTED = "<STR_LIT>" <EOL> _RESULTS = frozenset ( [ RESULT_SUCCESS , RESULT_FAILED , RESULT_ABORTED ] ) <EOL> _MAX_STATES_IN_MEMORY = <NUM_LIT:10> <EOL> mapreduce_id = db . StringProperty ( required = True ) <EOL> active = db . BooleanProperty ( default = True , indexed = False ) <EOL> input_finished = db . BooleanProperty ( default = False , indexed = False ) <EOL> counters_map = json_util . JsonProperty ( <EOL> CountersMap , default = CountersMap ( ) , indexed = False ) <EOL> result_status = db . StringProperty ( choices = _RESULTS , indexed = False ) <EOL> retries = db . IntegerProperty ( default = <NUM_LIT:0> , indexed = False ) <EOL> writer_state = json_util . JsonProperty ( dict , indexed = False ) <EOL> slice_id = db . IntegerProperty ( default = <NUM_LIT:0> , indexed = False ) <EOL> slice_start_time = db . DateTimeProperty ( indexed = False ) <EOL> slice_request_id = db . ByteStringProperty ( indexed = False ) <EOL> slice_retries = db . IntegerProperty ( default = <NUM_LIT:0> , indexed = False ) <EOL> acquired_once = db . BooleanProperty ( default = False , indexed = False ) <EOL> update_time = db . DateTimeProperty ( auto_now = True , indexed = False ) <EOL> shard_description = db . TextProperty ( default = "<STR_LIT>" ) <EOL> last_work_item = db . TextProperty ( default = "<STR_LIT>" ) <EOL> def __str__ ( self ) : <EOL> kv = { "<STR_LIT>" : self . active , <EOL> "<STR_LIT>" : self . slice_id , <EOL> "<STR_LIT>" : self . last_work_item , <EOL> "<STR_LIT>" : self . update_time } <EOL> if self . result_status : <EOL> kv [ "<STR_LIT>" ] = self . result_status <EOL> if self . retries : <EOL> kv [ "<STR_LIT>" ] = self . retries <EOL> if self . slice_start_time : <EOL> kv [ "<STR_LIT>" ] = self . slice_start_time <EOL> if self . slice_retries : <EOL> kv [ "<STR_LIT>" ] = self . slice_retries <EOL> if self . slice_request_id : <EOL> kv [ "<STR_LIT>" ] = self . slice_request_id <EOL> if self . acquired_once : <EOL> kv [ "<STR_LIT>" ] = self . acquired_once <EOL> keys = kv . keys ( ) <EOL> keys . sort ( ) <EOL> result = "<STR_LIT>" <EOL> for k in keys : <EOL> result += k + "<STR_LIT::>" + str ( kv [ k ] ) + "<STR_LIT:U+002C>" <EOL> result += "<STR_LIT:}>" <EOL> return result <EOL> def reset_for_retry ( self ) : <EOL> """<STR_LIT>""" <EOL> self . retries += <NUM_LIT:1> <EOL> self . last_work_item = "<STR_LIT>" <EOL> self . active = True <EOL> self . result_status = None <EOL> self . input_finished = False <EOL> self . counters_map = CountersMap ( ) <EOL> self . slice_id = <NUM_LIT:0> <EOL> self . slice_start_time = None <EOL> self . slice_request_id = None <EOL> self . slice_retries = <NUM_LIT:0> <EOL> self . acquired_once = False <EOL> def advance_for_next_slice ( self , recovery_slice = False ) : <EOL> """<STR_LIT>""" <EOL> self . slice_start_time = None <EOL> self . slice_request_id = None <EOL> self . slice_retries = <NUM_LIT:0> <EOL> self . acquired_once = False <EOL> if recovery_slice : <EOL> self . slice_id += <NUM_LIT:2> <EOL> else : <EOL> self . slice_id += <NUM_LIT:1> <EOL> def set_for_failure ( self ) : <EOL> self . active = False <EOL> self . result_status = self . RESULT_FAILED <EOL> def set_for_abort ( self ) : <EOL> self . active = False <EOL> self . result_status = self . RESULT_ABORTED <EOL> def set_input_finished ( self ) : <EOL> self . input_finished = True <EOL> def is_input_finished ( self ) : <EOL> return self . input_finished <EOL> def set_for_success ( self ) : <EOL> self . active = False <EOL> self . result_status = self . RESULT_SUCCESS <EOL> self . slice_start_time = None <EOL> self . slice_request_id = None <EOL> self . slice_retries = <NUM_LIT:0> <EOL> self . acquired_once = False <EOL> def copy_from ( self , other_state ) : <EOL> """<STR_LIT>""" <EOL> for prop in self . properties ( ) . values ( ) : <EOL> setattr ( self , prop . name , getattr ( other_state , prop . name ) ) <EOL> def __eq__ ( self , other ) : <EOL> if not isinstance ( other , self . __class__ ) : <EOL> return False <EOL> return self . properties ( ) == other . properties ( ) <EOL> def get_shard_number ( self ) : <EOL> """<STR_LIT>""" <EOL> return int ( self . key ( ) . name ( ) . split ( "<STR_LIT:->" ) [ - <NUM_LIT:1> ] ) <EOL> shard_number = property ( get_shard_number ) <EOL> def get_shard_id ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . key ( ) . name ( ) <EOL> shard_id = property ( get_shard_id ) <EOL> @ classmethod <EOL> def kind ( cls ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" <EOL> @ classmethod <EOL> def shard_id_from_number ( cls , mapreduce_id , shard_number ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( mapreduce_id , shard_number ) <EOL> @ classmethod <EOL> def get_key_by_shard_id ( cls , shard_id ) : <EOL> """<STR_LIT>""" <EOL> return db . Key . from_path ( cls . kind ( ) , shard_id ) <EOL> @ classmethod <EOL> def get_by_shard_id ( cls , shard_id ) : <EOL> """<STR_LIT>""" <EOL> return cls . get_by_key_name ( shard_id ) <EOL> @ classmethod <EOL> def find_by_mapreduce_state ( cls , mapreduce_state ) : <EOL> """<STR_LIT>""" <EOL> return list ( cls . find_all_by_mapreduce_state ( mapreduce_state ) ) <EOL> @ classmethod <EOL> def find_all_by_mapreduce_state ( cls , mapreduce_state ) : <EOL> """<STR_LIT>""" <EOL> keys = cls . calculate_keys_by_mapreduce_state ( mapreduce_state ) <EOL> i = <NUM_LIT:0> <EOL> while i < len ( keys ) : <EOL> @ db . non_transactional <EOL> def no_tx_get ( i ) : <EOL> return db . get ( keys [ i : i + cls . _MAX_STATES_IN_MEMORY ] ) <EOL> states = no_tx_get ( i ) <EOL> for s in states : <EOL> i += <NUM_LIT:1> <EOL> if s is not None : <EOL> yield s <EOL> @ classmethod <EOL> def calculate_keys_by_mapreduce_state ( cls , mapreduce_state ) : <EOL> """<STR_LIT>""" <EOL> if mapreduce_state is None : <EOL> return [ ] <EOL> keys = [ ] <EOL> for i in range ( mapreduce_state . mapreduce_spec . mapper . shard_count ) : <EOL> shard_id = cls . shard_id_from_number ( mapreduce_state . key ( ) . name ( ) , i ) <EOL> keys . append ( cls . get_key_by_shard_id ( shard_id ) ) <EOL> return keys <EOL> @ classmethod <EOL> def create_new ( cls , mapreduce_id , shard_number ) : <EOL> """<STR_LIT>""" <EOL> shard_id = cls . shard_id_from_number ( mapreduce_id , shard_number ) <EOL> state = cls ( key_name = shard_id , <EOL> mapreduce_id = mapreduce_id ) <EOL> return state <EOL> class MapreduceControl ( db . Model ) : <EOL> """<STR_LIT>""" <EOL> ABORT = "<STR_LIT>" <EOL> _COMMANDS = frozenset ( [ ABORT ] ) <EOL> _KEY_NAME = "<STR_LIT>" <EOL> command = db . TextProperty ( choices = _COMMANDS , required = True ) <EOL> @ classmethod <EOL> def kind ( cls ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" <EOL> @ classmethod <EOL> def get_key_by_job_id ( cls , mapreduce_id ) : <EOL> """<STR_LIT>""" <EOL> return db . Key . from_path ( cls . kind ( ) , "<STR_LIT>" % ( mapreduce_id , cls . _KEY_NAME ) ) <EOL> @ classmethod <EOL> def abort ( cls , mapreduce_id , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> cls ( key_name = "<STR_LIT>" % ( mapreduce_id , cls . _KEY_NAME ) , <EOL> command = cls . ABORT ) . put ( ** kwargs ) <EOL> class QuerySpec ( object ) : <EOL> """<STR_LIT>""" <EOL> DEFAULT_BATCH_SIZE = <NUM_LIT:50> <EOL> def __init__ ( self , <EOL> entity_kind , <EOL> keys_only = None , <EOL> filters = None , <EOL> batch_size = None , <EOL> model_class_path = None , <EOL> app = None , <EOL> ns = None ) : <EOL> self . entity_kind = entity_kind <EOL> self . keys_only = keys_only or False <EOL> self . filters = filters or None <EOL> self . batch_size = batch_size or self . DEFAULT_BATCH_SIZE <EOL> self . model_class_path = model_class_path <EOL> self . app = app <EOL> self . ns = ns <EOL> def to_json ( self ) : <EOL> return { "<STR_LIT>" : self . entity_kind , <EOL> "<STR_LIT>" : self . keys_only , <EOL> "<STR_LIT>" : self . filters , <EOL> "<STR_LIT>" : self . batch_size , <EOL> "<STR_LIT>" : self . model_class_path , <EOL> "<STR_LIT>" : self . app , <EOL> "<STR_LIT>" : self . ns } <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> return cls ( json [ "<STR_LIT>" ] , <EOL> json [ "<STR_LIT>" ] , <EOL> json [ "<STR_LIT>" ] , <EOL> json [ "<STR_LIT>" ] , <EOL> json [ "<STR_LIT>" ] , <EOL> json [ "<STR_LIT>" ] , <EOL> json [ "<STR_LIT>" ] ) </s>
