<s> from django . conf import settings <EOL> from django . db import router <EOL> from mapreduce . mapper_pipeline import MapperPipeline <EOL> from mapreduce import parameters <EOL> from mapreduce import control <EOL> from mapreduce import model <EOL> from pipeline . util import for_name <EOL> BASE_PATH = '<STR_LIT>' <EOL> PIPELINE_BASE_PATH = BASE_PATH + '<STR_LIT>' <EOL> class DjangaeMapperPipeline ( MapperPipeline ) : <EOL> def run ( self , job_name , handler_spec , input_reader_spec , output_writer_spec = None , params = None , shards = None ) : <EOL> """<STR_LIT>""" <EOL> if shards is None : <EOL> shards = parameters . config . SHARD_COUNT <EOL> mapreduce_id = control . start_map ( <EOL> job_name , <EOL> handler_spec , <EOL> input_reader_spec , <EOL> params or { } , <EOL> mapreduce_parameters = { <EOL> "<STR_LIT>" : self . get_callback_url ( ) , <EOL> "<STR_LIT>" : "<STR_LIT:GET>" , <EOL> "<STR_LIT>" : self . pipeline_id , <EOL> "<STR_LIT>" : BASE_PATH , <EOL> } , <EOL> shard_count = shards , <EOL> output_writer_spec = output_writer_spec , <EOL> queue_name = self . queue_name , <EOL> ) <EOL> self . fill ( self . outputs . job_id , mapreduce_id ) <EOL> self . set_status ( console_url = "<STR_LIT>" % ( <EOL> ( parameters . config . BASE_PATH , mapreduce_id ) ) ) <EOL> def callback ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( DjangaeMapperPipeline , self ) . callback ( ) <EOL> mapreduce_id = self . outputs . job_id . value <EOL> mapreduce_state = model . MapreduceState . get_by_job_id ( mapreduce_id ) <EOL> params = mapreduce_state . mapreduce_spec . mapper . params <EOL> finish_func = params . get ( '<STR_LIT>' , None ) <EOL> if not finish_func : <EOL> return None <EOL> finish_func = for_name ( finish_func ) <EOL> return finish_func ( * params . get ( '<STR_LIT:args>' , [ ] ) , ** params . get ( '<STR_LIT>' , { } ) ) <EOL> class MapReduceTask ( object ) : <EOL> """<STR_LIT>""" <EOL> shard_count = None <EOL> pipeline_class = MapperPipeline <EOL> job_name = None <EOL> queue_name = '<STR_LIT:default>' <EOL> output_writer_spec = None <EOL> mapreduce_parameters = { } <EOL> countdown = None <EOL> eta = None <EOL> model = None <EOL> map_args = [ ] <EOL> map_kwargs = { } <EOL> target = '<STR_LIT:default>' <EOL> def __init__ ( self , model = None , db = None ) : <EOL> if model : <EOL> self . model = model <EOL> self . db = db or router . db_for_read ( self . model ) <EOL> else : <EOL> self . db = db or '<STR_LIT:default>' <EOL> if not self . job_name : <EOL> self . job_name = self . get_class_path ( ) <EOL> def get_model_app_ ( self ) : <EOL> app = self . model . _meta . app_label <EOL> name = self . model . __name__ <EOL> return '<STR_LIT>' . format ( <EOL> app = app , <EOL> name = name , <EOL> ) <EOL> def get_class_path ( self ) : <EOL> return '<STR_LIT>' . format ( <EOL> mod = self . __class__ . __module__ , <EOL> cls = self . __class__ . __name__ , <EOL> ) <EOL> @ classmethod <EOL> def get_relative_path ( cls , func ) : <EOL> return '<STR_LIT>' . format ( <EOL> mod = cls . __module__ , <EOL> cls = cls . __name__ , <EOL> func = func . __name__ , <EOL> ) <EOL> @ staticmethod <EOL> def map ( entity , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def finish ( ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def start ( self , * args , ** kwargs ) : <EOL> queue_name = kwargs . pop ( '<STR_LIT>' , self . queue_name ) <EOL> mapper_parameters = { <EOL> '<STR_LIT>' : self . get_model_app_ ( ) , <EOL> '<STR_LIT>' : kwargs , <EOL> '<STR_LIT:args>' : args , <EOL> '<STR_LIT>' : settings . DATABASES . get ( self . db , { } ) . get ( "<STR_LIT>" ) , <EOL> '<STR_LIT>' : self . db <EOL> } <EOL> if '<STR_LIT>' not in self . __class__ . __dict__ : <EOL> raise Exception ( '<STR_LIT>' . format ( self . __class__ ) ) <EOL> mapper_parameters [ '<STR_LIT>' ] = self . get_relative_path ( self . map ) <EOL> if '<STR_LIT>' in self . __class__ . __dict__ : <EOL> mapper_parameters [ '<STR_LIT>' ] = self . get_relative_path ( self . finish ) <EOL> shard_count = self . shard_count or ( self . model . objects . using ( self . db ) . all ( ) [ : <NUM_LIT:1000> ] . count ( ) / <NUM_LIT:100> ) + <NUM_LIT:1> <EOL> pipe = DjangaeMapperPipeline ( <EOL> self . job_name , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> params = mapper_parameters , <EOL> shards = shard_count <EOL> ) <EOL> pipe . with_params ( target = self . target ) <EOL> pipe . start ( base_path = PIPELINE_BASE_PATH , queue_name = queue_name ) </s>
