<s> import base64 <EOL> from collections import defaultdict , namedtuple <EOL> from datetime import datetime , timedelta <EOL> import logging <EOL> import urllib <EOL> import urlparse <EOL> from requests . exceptions import Timeout , ConnectionError <EOL> from corehq . apps . cachehq . mixins import QuickCachedDocumentMixin <EOL> from corehq . form_processor . exceptions import XFormNotFound <EOL> from corehq . util . datadog . metrics import REPEATER_ERROR_COUNT <EOL> from corehq . util . datadog . utils import log_counter <EOL> from corehq . util . quickcache import quickcache <EOL> from dimagi . ext . couchdbkit import * <EOL> from couchdbkit . exceptions import ResourceNotFound <EOL> from django . core . cache import cache <EOL> import hashlib <EOL> from casexml . apps . case . xml import V2 , LEGAL_VERSIONS <EOL> from corehq . apps . receiverwrapper . exceptions import DuplicateFormatException , IgnoreDocument <EOL> from corehq . form_processor . interfaces . dbaccessors import FormAccessors , CaseAccessors <EOL> from couchforms . const import DEVICE_LOG_XMLNS <EOL> from dimagi . utils . decorators . memoized import memoized <EOL> from dimagi . utils . parsing import json_format_datetime <EOL> from dimagi . utils . mixins import UnicodeMixIn <EOL> from dimagi . utils . post import simple_post <EOL> from . dbaccessors import ( <EOL> get_pending_repeat_record_count , <EOL> get_failure_repeat_record_count , <EOL> get_success_repeat_record_count , <EOL> ) <EOL> from . const import ( <EOL> MAX_RETRY_WAIT , <EOL> MIN_RETRY_WAIT , <EOL> RECORD_FAILURE_STATE , <EOL> RECORD_SUCCESS_STATE , <EOL> RECORD_PENDING_STATE , <EOL> POST_TIMEOUT , <EOL> ) <EOL> from . exceptions import RequestConnectionError <EOL> repeater_types = { } <EOL> def register_repeater_type ( cls ) : <EOL> repeater_types [ cls . __name__ ] = cls <EOL> return cls <EOL> def simple_post_with_cached_timeout ( data , url , expiry = <NUM_LIT> * <NUM_LIT> , force_send = False , * args , ** kwargs ) : <EOL> key = hashlib . md5 ( <EOL> '<STR_LIT>' . format ( __name__ , url ) <EOL> ) . hexdigest ( ) <EOL> cache_value = cache . get ( key ) <EOL> if cache_value and not force_send : <EOL> raise RequestConnectionError ( cache_value ) <EOL> try : <EOL> resp = simple_post ( data , url , * args , ** kwargs ) <EOL> except ( Timeout , ConnectionError ) , e : <EOL> cache . set ( key , e . message , expiry ) <EOL> raise RequestConnectionError ( e . message ) <EOL> if not <NUM_LIT:200> <= resp . status_code < <NUM_LIT> : <EOL> message = u'<STR_LIT>' . format ( resp . status_code , resp . reason ) <EOL> cache . set ( key , message , expiry ) <EOL> raise RequestConnectionError ( message ) <EOL> return resp <EOL> DELETED = "<STR_LIT>" <EOL> FormatInfo = namedtuple ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> class GeneratorCollection ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , repeater_class ) : <EOL> self . repeater_class = repeater_class <EOL> self . default_format = '<STR_LIT>' <EOL> self . format_generator_map = { } <EOL> def add_new_format ( self , format_name , format_label , generator_class , is_default = False ) : <EOL> """<STR_LIT>""" <EOL> if is_default and self . default_format : <EOL> raise DuplicateFormatException ( "<STR_LIT>" ) <EOL> elif is_default : <EOL> self . default_format = format_name <EOL> if format_name in self . format_generator_map : <EOL> raise DuplicateFormatException ( "<STR_LIT>" ) <EOL> self . format_generator_map [ format_name ] = FormatInfo ( <EOL> name = format_name , <EOL> label = format_label , <EOL> generator_class = generator_class <EOL> ) <EOL> def get_default_format ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . default_format <EOL> def get_default_generator ( self ) : <EOL> """<STR_LIT>""" <EOL> raise self . format_generator_map [ self . default_format ] . generator_class <EOL> def get_all_formats ( self , for_domain = None ) : <EOL> """<STR_LIT>""" <EOL> return [ ( name , format . label ) for name , format in self . format_generator_map . iteritems ( ) <EOL> if not for_domain or format . generator_class . enabled_for_domain ( for_domain ) ] <EOL> def get_generator_by_format ( self , format ) : <EOL> """<STR_LIT>""" <EOL> return self . format_generator_map [ format ] . generator_class <EOL> class RegisterGenerator ( object ) : <EOL> """<STR_LIT>""" <EOL> generators = { } <EOL> def __init__ ( self , repeater_cls , format_name , format_label , is_default = False ) : <EOL> self . format_name = format_name <EOL> self . format_label = format_label <EOL> self . repeater_cls = repeater_cls <EOL> self . label = format_label <EOL> self . is_default = is_default <EOL> def __call__ ( self , generator_class ) : <EOL> if not self . repeater_cls in RegisterGenerator . generators : <EOL> RegisterGenerator . generators [ self . repeater_cls ] = GeneratorCollection ( self . repeater_cls ) <EOL> RegisterGenerator . generators [ self . repeater_cls ] . add_new_format ( <EOL> self . format_name , <EOL> self . format_label , <EOL> generator_class , <EOL> is_default = self . is_default <EOL> ) <EOL> return generator_class <EOL> @ classmethod <EOL> def generator_class_by_repeater_format ( cls , repeater_class , format_name ) : <EOL> """<STR_LIT>""" <EOL> generator_collection = cls . generators [ repeater_class ] <EOL> return generator_collection . get_generator_by_format ( format_name ) <EOL> @ classmethod <EOL> def all_formats_by_repeater ( cls , repeater_class , for_domain = None ) : <EOL> """<STR_LIT>""" <EOL> generator_collection = cls . generators [ repeater_class ] <EOL> return generator_collection . get_all_formats ( for_domain = for_domain ) <EOL> @ classmethod <EOL> def default_format_by_repeater ( cls , repeater_class ) : <EOL> """<STR_LIT>""" <EOL> generator_collection = cls . generators [ repeater_class ] <EOL> return generator_collection . get_default_format ( ) <EOL> class Repeater ( QuickCachedDocumentMixin , Document , UnicodeMixIn ) : <EOL> """<STR_LIT>""" <EOL> base_doc = '<STR_LIT>' <EOL> domain = StringProperty ( ) <EOL> url = StringProperty ( ) <EOL> format = StringProperty ( ) <EOL> use_basic_auth = BooleanProperty ( default = False ) <EOL> username = StringProperty ( ) <EOL> password = StringProperty ( ) <EOL> def get_pending_record_count ( self ) : <EOL> return get_pending_repeat_record_count ( self . domain , self . _id ) <EOL> def get_failure_record_count ( self ) : <EOL> return get_failure_repeat_record_count ( self . domain , self . _id ) <EOL> def get_success_record_count ( self ) : <EOL> return get_success_repeat_record_count ( self . domain , self . _id ) <EOL> def format_or_default_format ( self ) : <EOL> return self . format or RegisterGenerator . default_format_by_repeater ( self . __class__ ) <EOL> def get_payload_generator ( self , payload_format ) : <EOL> gen = RegisterGenerator . generator_class_by_repeater_format ( self . __class__ , payload_format ) <EOL> return gen ( self ) <EOL> def payload_doc ( self , repeat_record ) : <EOL> raise NotImplementedError <EOL> def get_payload ( self , repeat_record ) : <EOL> generator = self . get_payload_generator ( self . format_or_default_format ( ) ) <EOL> return generator . get_payload ( repeat_record , self . payload_doc ( repeat_record ) ) <EOL> def register ( self , payload , next_check = None ) : <EOL> if not self . allowed_to_forward ( payload ) : <EOL> return <EOL> repeat_record = RepeatRecord ( <EOL> repeater_id = self . get_id , <EOL> repeater_type = self . doc_type , <EOL> domain = self . domain , <EOL> next_check = next_check or datetime . utcnow ( ) , <EOL> payload_id = payload . get_id <EOL> ) <EOL> repeat_record . save ( ) <EOL> return repeat_record <EOL> def allowed_to_forward ( self , payload ) : <EOL> """<STR_LIT>""" <EOL> return True <EOL> def clear_caches ( self ) : <EOL> if self . __class__ == Repeater : <EOL> cls = self . get_class_from_doc_type ( self . doc_type ) <EOL> else : <EOL> cls = self . __class__ <EOL> Repeater . by_domain . clear ( cls , self . domain ) <EOL> Repeater . by_domain . clear ( Repeater , self . domain ) <EOL> @ classmethod <EOL> @ quickcache ( [ '<STR_LIT>' , '<STR_LIT>' ] , timeout = <NUM_LIT:5> * <NUM_LIT> , memoize_timeout = <NUM_LIT:10> ) <EOL> def by_domain ( cls , domain ) : <EOL> key = [ domain ] <EOL> if cls . __name__ in repeater_types : <EOL> key . append ( cls . __name__ ) <EOL> elif cls . __name__ == Repeater . __name__ : <EOL> pass <EOL> else : <EOL> return [ ] <EOL> raw_docs = cls . view ( '<STR_LIT>' , <EOL> startkey = key , <EOL> endkey = key + [ { } ] , <EOL> include_docs = True , <EOL> reduce = False , <EOL> wrap_doc = False <EOL> ) <EOL> return [ cls . wrap ( repeater_doc [ '<STR_LIT>' ] ) for repeater_doc in raw_docs <EOL> if cls . get_class_from_doc_type ( repeater_doc [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) ] <EOL> @ classmethod <EOL> def wrap ( cls , data ) : <EOL> if cls . __name__ == Repeater . __name__ : <EOL> cls_ = cls . get_class_from_doc_type ( data [ '<STR_LIT>' ] ) <EOL> if cls_ : <EOL> return cls_ . wrap ( data ) <EOL> else : <EOL> raise ResourceNotFound ( '<STR_LIT>' % data ) <EOL> else : <EOL> return super ( Repeater , cls ) . wrap ( data ) <EOL> @ staticmethod <EOL> def get_class_from_doc_type ( doc_type ) : <EOL> doc_type = doc_type . replace ( DELETED , '<STR_LIT>' ) <EOL> if doc_type in repeater_types : <EOL> return repeater_types [ doc_type ] <EOL> else : <EOL> return None <EOL> def retire ( self ) : <EOL> if DELETED not in self [ '<STR_LIT>' ] : <EOL> self [ '<STR_LIT>' ] += DELETED <EOL> if DELETED not in self [ '<STR_LIT>' ] : <EOL> self [ '<STR_LIT>' ] += DELETED <EOL> self . save ( ) <EOL> def get_url ( self , repeate_record ) : <EOL> return self . url <EOL> def get_headers ( self , repeat_record ) : <EOL> generator = self . get_payload_generator ( self . format_or_default_format ( ) ) <EOL> headers = generator . get_headers ( ) <EOL> if self . use_basic_auth : <EOL> user_pass = base64 . encodestring ( '<STR_LIT::>' . join ( ( self . username , self . password ) ) ) . replace ( '<STR_LIT:\n>' , '<STR_LIT>' ) <EOL> headers . update ( { '<STR_LIT>' : '<STR_LIT>' + user_pass } ) <EOL> return headers <EOL> @ register_repeater_type <EOL> class FormRepeater ( Repeater ) : <EOL> """<STR_LIT>""" <EOL> include_app_id_param = BooleanProperty ( default = True ) <EOL> @ memoized <EOL> def payload_doc ( self , repeat_record ) : <EOL> return FormAccessors ( repeat_record . domain ) . get_form ( repeat_record . payload_id ) <EOL> def allowed_to_forward ( self , payload ) : <EOL> return payload . xmlns != DEVICE_LOG_XMLNS <EOL> def get_url ( self , repeat_record ) : <EOL> url = super ( FormRepeater , self ) . get_url ( repeat_record ) <EOL> if not self . include_app_id_param : <EOL> return url <EOL> else : <EOL> url_parts = list ( urlparse . urlparse ( url ) ) <EOL> query = urlparse . parse_qsl ( url_parts [ <NUM_LIT:4> ] ) <EOL> query . append ( ( "<STR_LIT>" , self . payload_doc ( repeat_record ) . app_id ) ) <EOL> url_parts [ <NUM_LIT:4> ] = urllib . urlencode ( query ) <EOL> return urlparse . urlunparse ( url_parts ) <EOL> def get_headers ( self , repeat_record ) : <EOL> headers = super ( FormRepeater , self ) . get_headers ( repeat_record ) <EOL> headers . update ( { <EOL> "<STR_LIT>" : self . payload_doc ( repeat_record ) . received_on . isoformat ( ) + "<STR_LIT>" <EOL> } ) <EOL> return headers <EOL> def __unicode__ ( self ) : <EOL> return "<STR_LIT>" % self . url <EOL> @ register_repeater_type <EOL> class CaseRepeater ( Repeater ) : <EOL> """<STR_LIT>""" <EOL> version = StringProperty ( default = V2 , choices = LEGAL_VERSIONS ) <EOL> white_listed_case_types = StringListProperty ( default = [ ] ) <EOL> black_listed_users = StringListProperty ( default = [ ] ) <EOL> def allowed_to_forward ( self , payload ) : <EOL> allowed_case_type = not self . white_listed_case_types or payload . type in self . white_listed_case_types <EOL> allowed_user = self . payload_user_id ( payload ) not in self . black_listed_users <EOL> return allowed_case_type and allowed_user <EOL> def payload_user_id ( self , payload ) : <EOL> return payload . actions [ - <NUM_LIT:1> ] . user_id <EOL> @ memoized <EOL> def payload_doc ( self , repeat_record ) : <EOL> return CaseAccessors ( repeat_record . domain ) . get_case ( repeat_record . payload_id ) <EOL> def get_headers ( self , repeat_record ) : <EOL> headers = super ( CaseRepeater , self ) . get_headers ( repeat_record ) <EOL> headers . update ( { <EOL> "<STR_LIT>" : self . payload_doc ( repeat_record ) . server_modified_on . isoformat ( ) + "<STR_LIT>" <EOL> } ) <EOL> return headers <EOL> def __unicode__ ( self ) : <EOL> return "<STR_LIT>" % self . url <EOL> @ register_repeater_type <EOL> class ShortFormRepeater ( Repeater ) : <EOL> """<STR_LIT>""" <EOL> version = StringProperty ( default = V2 , choices = LEGAL_VERSIONS ) <EOL> @ memoized <EOL> def payload_doc ( self , repeat_record ) : <EOL> return FormAccessors ( repeat_record . domain ) . get_form ( repeat_record . payload_id ) <EOL> def allowed_to_forward ( self , payload ) : <EOL> return payload . xmlns != DEVICE_LOG_XMLNS <EOL> def get_headers ( self , repeat_record ) : <EOL> headers = super ( ShortFormRepeater , self ) . get_headers ( repeat_record ) <EOL> headers . update ( { <EOL> "<STR_LIT>" : self . payload_doc ( repeat_record ) . received_on . isoformat ( ) + "<STR_LIT>" <EOL> } ) <EOL> return headers <EOL> def __unicode__ ( self ) : <EOL> return "<STR_LIT>" % self . url <EOL> @ register_repeater_type <EOL> class AppStructureRepeater ( Repeater ) : <EOL> def payload_doc ( self , repeat_record ) : <EOL> return None <EOL> class RepeatRecord ( Document ) : <EOL> """<STR_LIT>""" <EOL> repeater_id = StringProperty ( ) <EOL> repeater_type = StringProperty ( ) <EOL> domain = StringProperty ( ) <EOL> last_checked = DateTimeProperty ( ) <EOL> next_check = DateTimeProperty ( ) <EOL> succeeded = BooleanProperty ( default = False ) <EOL> failure_reason = StringProperty ( ) <EOL> payload_id = StringProperty ( ) <EOL> @ property <EOL> @ memoized <EOL> def repeater ( self ) : <EOL> return Repeater . get ( self . repeater_id ) <EOL> @ property <EOL> def url ( self ) : <EOL> try : <EOL> return self . repeater . get_url ( self ) <EOL> except XFormNotFound : <EOL> return None <EOL> @ property <EOL> def state ( self ) : <EOL> state = RECORD_PENDING_STATE <EOL> if self . succeeded : <EOL> state = RECORD_SUCCESS_STATE <EOL> elif self . failure_reason : <EOL> state = RECORD_FAILURE_STATE <EOL> return state <EOL> @ classmethod <EOL> def all ( cls , domain = None , due_before = None , limit = None ) : <EOL> json_now = json_format_datetime ( due_before or datetime . utcnow ( ) ) <EOL> repeat_records = RepeatRecord . view ( "<STR_LIT>" , <EOL> startkey = [ domain ] , <EOL> endkey = [ domain , json_now , { } ] , <EOL> include_docs = True , <EOL> reduce = False , <EOL> limit = limit , <EOL> ) <EOL> return repeat_records <EOL> @ classmethod <EOL> def count ( cls , domain = None ) : <EOL> results = RepeatRecord . view ( "<STR_LIT>" , <EOL> startkey = [ domain ] , <EOL> endkey = [ domain , { } ] , <EOL> reduce = True , <EOL> ) . one ( ) <EOL> return results [ '<STR_LIT:value>' ] if results else <NUM_LIT:0> <EOL> def update_success ( self ) : <EOL> self . last_checked = datetime . utcnow ( ) <EOL> self . next_check = None <EOL> self . succeeded = True <EOL> def update_failure ( self , reason = None ) : <EOL> assert self . succeeded is False <EOL> assert self . next_check is not None <EOL> now = datetime . utcnow ( ) <EOL> window = timedelta ( minutes = <NUM_LIT:0> ) <EOL> if self . last_checked : <EOL> window = self . next_check - self . last_checked <EOL> window += ( window // <NUM_LIT:2> ) <EOL> if window < MIN_RETRY_WAIT : <EOL> window = MIN_RETRY_WAIT <EOL> elif window > MAX_RETRY_WAIT : <EOL> window = MAX_RETRY_WAIT <EOL> self . last_checked = now <EOL> self . next_check = self . last_checked + window <EOL> self . failure_reason = reason <EOL> def try_now ( self ) : <EOL> return not self . succeeded <EOL> def get_payload ( self ) : <EOL> return self . repeater . get_payload ( self ) <EOL> def fire ( self , max_tries = <NUM_LIT:3> , force_send = False ) : <EOL> try : <EOL> payload = self . get_payload ( ) <EOL> except ResourceNotFound : <EOL> logging . exception ( u'<STR_LIT>' . format ( <EOL> self . _id , self . domain , <EOL> ) ) <EOL> self . doc_type = self . doc_type + '<STR_LIT>' <EOL> self . save ( ) <EOL> except IgnoreDocument : <EOL> logging . info ( u'<STR_LIT>' . format ( <EOL> self . _id , self . domain , <EOL> ) ) <EOL> self . update_success ( ) <EOL> else : <EOL> headers = self . repeater . get_headers ( self ) <EOL> if self . try_now ( ) or force_send : <EOL> failure_reason = None <EOL> for i in range ( max_tries ) : <EOL> try : <EOL> resp = simple_post_with_cached_timeout ( <EOL> payload , <EOL> self . url , <EOL> headers = headers , <EOL> force_send = force_send , <EOL> timeout = POST_TIMEOUT , <EOL> ) <EOL> if <NUM_LIT:200> <= resp . status_code < <NUM_LIT> : <EOL> self . update_success ( ) <EOL> break <EOL> else : <EOL> failure_reason = u'<STR_LIT>' . format ( resp . status_code , resp . reason ) <EOL> except Exception , e : <EOL> failure_reason = unicode ( e ) <EOL> if not self . succeeded : <EOL> self . update_failure ( failure_reason ) <EOL> log_counter ( REPEATER_ERROR_COUNT , { <EOL> '<STR_LIT>' : self . _id , <EOL> '<STR_LIT>' : failure_reason , <EOL> '<STR_LIT>' : self . url , <EOL> } ) <EOL> from corehq . apps . repeaters import signals </s>
