<s> """<STR_LIT>""" <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 model <EOL> from google . appengine . ext . mapreduce import util <EOL> from google . appengine . ext . mapreduce . api . map_job import map_job_config <EOL> class Job ( object ) : <EOL> """<STR_LIT>""" <EOL> RUNNING = "<STR_LIT>" <EOL> FAILED = model . MapreduceState . RESULT_FAILED <EOL> ABORTED = model . MapreduceState . RESULT_ABORTED <EOL> SUCCESS = model . MapreduceState . RESULT_SUCCESS <EOL> STATUS_ENUM = [ RUNNING , FAILED , ABORTED , SUCCESS ] <EOL> def __init__ ( self , state = None ) : <EOL> """<STR_LIT>""" <EOL> self . _state = state <EOL> self . job_config = map_job_config . JobConfig . _to_map_job_config ( <EOL> state . mapreduce_spec , <EOL> queue_name = state . mapreduce_spec . params . get ( "<STR_LIT>" ) ) <EOL> @ classmethod <EOL> def get_job_by_id ( cls , job_id = None ) : <EOL> """<STR_LIT>""" <EOL> state = cls . __get_state_by_id ( job_id ) <EOL> return cls ( state ) <EOL> def get_status ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __update_state ( ) <EOL> if self . _state . active : <EOL> return self . RUNNING <EOL> else : <EOL> return self . _state . result_status <EOL> def abort ( self ) : <EOL> """<STR_LIT>""" <EOL> model . MapreduceControl . abort ( self . job_config . job_id ) <EOL> def get_counters ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __update_state ( ) <EOL> return self . _state . counters_map . counters . iteritems ( ) <EOL> def get_counter ( self , counter_name , default = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> self . __update_state ( ) <EOL> return self . _state . counters_map . get ( counter_name , default ) <EOL> def get_outputs ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . SUCCESS == self . get_status ( ) <EOL> ss = model . ShardState . find_all_by_mapreduce_state ( self . _state ) <EOL> for s in ss : <EOL> yield iter ( s . writer_state . get ( "<STR_LIT>" , [ ] ) ) <EOL> @ classmethod <EOL> def submit ( cls , job_config , in_xg_transaction = False ) : <EOL> """<STR_LIT>""" <EOL> cls . __validate_job_config ( job_config ) <EOL> mapper_spec = job_config . _get_mapper_spec ( ) <EOL> mapreduce_params = job_config . _get_mr_params ( ) <EOL> mapreduce_spec = model . MapreduceSpec ( <EOL> job_config . job_name , <EOL> job_config . job_id , <EOL> mapper_spec . to_json ( ) , <EOL> mapreduce_params , <EOL> util . _obj_to_path ( job_config . _hooks_cls ) ) <EOL> if in_xg_transaction : <EOL> propagation = db . MANDATORY <EOL> else : <EOL> propagation = db . INDEPENDENT <EOL> state = None <EOL> @ db . transactional ( propagation = propagation ) <EOL> def _txn ( ) : <EOL> state = cls . __create_and_save_state ( job_config , mapreduce_spec ) <EOL> cls . __add_kickoff_task ( job_config , mapreduce_spec ) <EOL> return state <EOL> state = _txn ( ) <EOL> return cls ( state ) <EOL> def __update_state ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _state . active : <EOL> self . _state = self . __get_state_by_id ( self . job_config . job_id ) <EOL> @ classmethod <EOL> def __get_state_by_id ( cls , job_id ) : <EOL> """<STR_LIT>""" <EOL> state = model . MapreduceState . get_by_job_id ( job_id ) <EOL> if state is None : <EOL> raise ValueError ( "<STR_LIT>" % job_id ) <EOL> return state <EOL> @ classmethod <EOL> def __validate_job_config ( cls , job_config ) : <EOL> job_config . input_reader_cls . validate ( job_config ) <EOL> if job_config . output_writer_cls : <EOL> job_config . output_writer_cls . validate ( job_config . _get_mapper_spec ( ) ) <EOL> @ classmethod <EOL> def __create_and_save_state ( cls , job_config , mapreduce_spec ) : <EOL> """<STR_LIT>""" <EOL> state = model . MapreduceState . create_new ( job_config . job_id ) <EOL> state . mapreduce_spec = mapreduce_spec <EOL> state . active = True <EOL> state . active_shards = <NUM_LIT:0> <EOL> state . app_id = job_config . _app <EOL> config = datastore_rpc . Configuration ( force_writes = job_config . _force_writes ) <EOL> state . put ( config = config ) <EOL> return state <EOL> @ classmethod <EOL> def __add_kickoff_task ( cls , job_config , mapreduce_spec ) : <EOL> """<STR_LIT>""" <EOL> params = { "<STR_LIT>" : job_config . job_id } <EOL> kickoff_task = taskqueue . Task ( <EOL> url = job_config . _base_path + "<STR_LIT>" + job_config . job_id , <EOL> headers = util . _get_task_headers ( job_config . job_id ) , <EOL> params = params ) <EOL> if job_config . _hooks_cls : <EOL> hooks = job_config . _hooks_cls ( mapreduce_spec ) <EOL> try : <EOL> hooks . enqueue_kickoff_task ( kickoff_task , job_config . queue_name ) <EOL> return <EOL> except NotImplementedError : <EOL> pass <EOL> kickoff_task . add ( job_config . queue_name , transactional = True ) </s>
