<s> from __future__ import division <EOL> from collections import namedtuple <EOL> from datetime import datetime <EOL> import importlib <EOL> import sys <EOL> import simplejson <EOL> from django . conf import settings <EOL> from dimagi . utils . chunked import chunked <EOL> from dimagi . utils . modules import to_function <EOL> from dimagi . utils . parsing import string_to_utc_datetime <EOL> from pillowtop . exceptions import PillowNotFoundError <EOL> def get_pillow_instance ( full_class_str ) : <EOL> pillow_class = _import_class_or_function ( full_class_str ) <EOL> return pillow_class ( ) <EOL> def _import_class_or_function ( full_class_str ) : <EOL> return to_function ( full_class_str , failhard = settings . DEBUG ) <EOL> def get_all_pillow_classes ( ) : <EOL> return [ config . get_class ( ) for config in get_all_pillow_configs ( ) ] <EOL> def get_all_pillow_instances ( ) : <EOL> return [ config . get_instance ( ) for config in get_all_pillow_configs ( ) ] <EOL> def get_all_pillow_configs ( ) : <EOL> return get_pillow_configs_from_settings_dict ( getattr ( settings , '<STR_LIT>' , { } ) ) <EOL> def get_pillow_configs_from_settings_dict ( pillow_settings_dict ) : <EOL> """<STR_LIT>""" <EOL> for section , list_of_pillows in pillow_settings_dict . items ( ) : <EOL> for pillow_config in list_of_pillows : <EOL> yield get_pillow_config_from_setting ( section , pillow_config ) <EOL> class PillowConfig ( namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT>' ] ) ) : <EOL> """<STR_LIT>""" <EOL> def get_class ( self ) : <EOL> return _import_class_or_function ( self . class_name ) <EOL> def get_instance ( self ) : <EOL> if self . instance_generator : <EOL> instance_generator_fn = _import_class_or_function ( self . instance_generator ) <EOL> return instance_generator_fn ( self . name ) <EOL> else : <EOL> return get_pillow_instance ( self . class_name ) <EOL> def get_pillow_config_from_setting ( section , pillow_config_string_or_dict ) : <EOL> if isinstance ( pillow_config_string_or_dict , basestring ) : <EOL> return PillowConfig ( <EOL> section , <EOL> pillow_config_string_or_dict . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) [ <NUM_LIT:1> ] , <EOL> pillow_config_string_or_dict , <EOL> None , <EOL> ) <EOL> else : <EOL> assert '<STR_LIT:class>' in pillow_config_string_or_dict <EOL> class_name = pillow_config_string_or_dict [ '<STR_LIT:class>' ] <EOL> return PillowConfig ( <EOL> section , <EOL> pillow_config_string_or_dict . get ( '<STR_LIT:name>' , class_name ) , <EOL> class_name , <EOL> pillow_config_string_or_dict . get ( '<STR_LIT>' , None ) , <EOL> ) <EOL> def get_pillow_by_name ( pillow_class_name , instantiate = True ) : <EOL> config = get_pillow_config_by_name ( pillow_class_name ) <EOL> return config . get_instance ( ) if instantiate else config . get_class ( ) <EOL> def get_pillow_config_by_name ( pillow_name ) : <EOL> all_configs = get_all_pillow_configs ( ) <EOL> for config in all_configs : <EOL> if config . name == pillow_name : <EOL> return config <EOL> raise PillowNotFoundError ( u'<STR_LIT>' . format ( pillow_name ) ) <EOL> def force_seq_int ( seq ) : <EOL> if seq is None or seq == '<STR_LIT>' : <EOL> return None <EOL> elif isinstance ( seq , dict ) : <EOL> return None <EOL> elif isinstance ( seq , basestring ) : <EOL> return int ( seq . split ( '<STR_LIT:->' ) [ <NUM_LIT:0> ] ) <EOL> else : <EOL> assert isinstance ( seq , int ) <EOL> return seq <EOL> def get_all_pillows_json ( ) : <EOL> pillow_configs = get_all_pillow_configs ( ) <EOL> return [ get_pillow_json ( pillow_config ) for pillow_config in pillow_configs ] <EOL> def get_pillow_json ( pillow_config ) : <EOL> assert isinstance ( pillow_config , PillowConfig ) <EOL> from pillowtop . listener import AliasedElasticPillow <EOL> pillow_class = pillow_config . get_class ( ) <EOL> pillow = ( pillow_class ( online = False ) if issubclass ( pillow_class , AliasedElasticPillow ) <EOL> else pillow_config . get_instance ( ) ) <EOL> checkpoint = pillow . get_checkpoint ( ) <EOL> timestamp = checkpoint . timestamp <EOL> if timestamp : <EOL> time_since_last = datetime . utcnow ( ) - timestamp <EOL> hours_since_last = time_since_last . total_seconds ( ) // <NUM_LIT> <EOL> try : <EOL> time_since_last = str ( time_since_last ) <EOL> time_since_last = time_since_last [ <NUM_LIT:0> : time_since_last . index ( '<STR_LIT:.>' ) ] <EOL> except ValueError : <EOL> pass <EOL> else : <EOL> time_since_last = '<STR_LIT>' <EOL> hours_since_last = None <EOL> try : <EOL> db_seq = pillow . get_change_feed ( ) . get_latest_change_id ( ) <EOL> except ValueError : <EOL> db_seq = None <EOL> return { <EOL> '<STR_LIT:name>' : pillow_config . name , <EOL> '<STR_LIT>' : force_seq_int ( checkpoint . wrapped_sequence ) , <EOL> '<STR_LIT>' : force_seq_int ( checkpoint . old_sequence ) or <NUM_LIT:0> , <EOL> '<STR_LIT>' : force_seq_int ( db_seq ) , <EOL> '<STR_LIT>' : time_since_last , <EOL> '<STR_LIT>' : hours_since_last <EOL> } <EOL> def prepare_bulk_payloads ( bulk_changes , max_size , chunk_size = <NUM_LIT:100> ) : <EOL> payloads = [ '<STR_LIT>' ] <EOL> for bulk_chunk in chunked ( bulk_changes , chunk_size ) : <EOL> current_payload = payloads [ - <NUM_LIT:1> ] <EOL> payload_chunk = '<STR_LIT:\n>' . join ( map ( simplejson . dumps , bulk_chunk ) ) + '<STR_LIT:\n>' <EOL> appended_payload = current_payload + payload_chunk <EOL> new_payload_size = sys . getsizeof ( appended_payload ) <EOL> if new_payload_size > max_size : <EOL> payloads . append ( payload_chunk ) <EOL> else : <EOL> payloads [ - <NUM_LIT:1> ] = appended_payload <EOL> return filter ( None , payloads ) </s>
