<s> from __future__ import absolute_import <EOL> from io import FileIO <EOL> import os <EOL> from uuid import uuid4 <EOL> import shutil <EOL> import hashlib <EOL> from couchdbkit import ResourceConflict , ResourceNotFound <EOL> from casexml . apps . phone . cache_utils import copy_payload_and_synclog_and_get_new_file <EOL> from casexml . apps . phone . data_providers import get_restore_providers , get_long_running_providers <EOL> from casexml . apps . phone . data_providers . case . load_testing import get_loadtest_factor <EOL> from casexml . apps . phone . exceptions import ( <EOL> MissingSyncLog , InvalidSyncLogException , SyncLogUserMismatch , <EOL> BadStateException , RestoreException , <EOL> ) <EOL> from corehq . toggles import LOOSE_SYNC_TOKEN_VALIDATION , EXTENSION_CASES_SYNC_ENABLED <EOL> from corehq . util . soft_assert import soft_assert <EOL> from dimagi . utils . decorators . memoized import memoized <EOL> from casexml . apps . phone . models import SyncLog , get_properly_wrapped_sync_log , LOG_FORMAT_SIMPLIFIED , get_sync_log_class_by_format <EOL> import logging <EOL> from dimagi . utils . couch . database import get_db , get_safe_write_kwargs <EOL> from casexml . apps . phone import xml as xml_util <EOL> from datetime import datetime , timedelta <EOL> from dimagi . utils . couch . cache . cache_core import get_redis_default_cache <EOL> from couchforms . openrosa_response import ( <EOL> ResponseNature , <EOL> get_simple_response_xml , <EOL> ) <EOL> from casexml . apps . case . xml import check_version , V1 <EOL> from django . http import HttpResponse , StreamingHttpResponse <EOL> from django . conf import settings <EOL> from casexml . apps . phone . checksum import CaseStateHash <EOL> from wsgiref . util import FileWrapper <EOL> from xml . etree import ElementTree <EOL> logger = logging . getLogger ( __name__ ) <EOL> INITIAL_SYNC_CACHE_TIMEOUT = <NUM_LIT> * <NUM_LIT> <EOL> INITIAL_SYNC_CACHE_THRESHOLD = <NUM_LIT> <EOL> def stream_response ( payload , headers = None ) : <EOL> try : <EOL> response = StreamingHttpResponse ( FileWrapper ( payload ) , content_type = "<STR_LIT>" ) <EOL> if headers : <EOL> for header , value in headers . items ( ) : <EOL> response [ header ] = value <EOL> return response <EOL> except IOError as e : <EOL> return HttpResponse ( e , status = <NUM_LIT> ) <EOL> class StockSettings ( object ) : <EOL> def __init__ ( self , section_to_consumption_types = None , consumption_config = None , <EOL> default_product_list = None , force_consumption_case_filter = None ) : <EOL> """<STR_LIT>""" <EOL> self . section_to_consumption_types = section_to_consumption_types or { } <EOL> self . consumption_config = consumption_config <EOL> self . default_product_list = default_product_list or [ ] <EOL> self . force_consumption_case_filter = force_consumption_case_filter or ( lambda stub : False ) <EOL> class RestoreResponse ( object ) : <EOL> start_tag_template = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) <EOL> items_template = '<STR_LIT>' <EOL> closing_tag = '<STR_LIT>' <EOL> def __init__ ( self , username = None , items = False ) : <EOL> self . username = username <EOL> self . items = items <EOL> self . num_items = <NUM_LIT:0> <EOL> self . finalized = False <EOL> def close ( self ) : <EOL> self . response_body . close ( ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_val , exc_tb ) : <EOL> self . close ( ) <EOL> def append ( self , xml_element ) : <EOL> self . num_items += <NUM_LIT:1> <EOL> if isinstance ( xml_element , basestring ) : <EOL> self . response_body . write ( xml_element ) <EOL> else : <EOL> self . response_body . write ( xml_util . tostring ( xml_element ) ) <EOL> def extend ( self , iterable ) : <EOL> for element in iterable : <EOL> self . append ( element ) <EOL> def finalize ( self ) : <EOL> raise NotImplemented ( ) <EOL> def get_cache_payload ( self , full = False ) : <EOL> raise NotImplemented ( ) <EOL> def as_string ( self ) : <EOL> raise NotImplemented ( ) <EOL> def __str__ ( self ) : <EOL> return self . as_string ( ) <EOL> class FileRestoreResponse ( RestoreResponse ) : <EOL> BODY_TAG_SUFFIX = '<STR_LIT>' <EOL> EXTENSION = '<STR_LIT>' <EOL> def __init__ ( self , username = None , items = False ) : <EOL> super ( FileRestoreResponse , self ) . __init__ ( username , items ) <EOL> self . filename = os . path . join ( settings . SHARED_DRIVE_CONF . restore_dir , uuid4 ( ) . hex ) <EOL> self . response_body = FileIO ( self . get_filename ( self . BODY_TAG_SUFFIX ) , '<STR_LIT>' ) <EOL> def get_filename ( self , suffix = None ) : <EOL> return "<STR_LIT>" . format ( <EOL> filename = self . filename , <EOL> suffix = suffix or '<STR_LIT>' , <EOL> ext = self . EXTENSION <EOL> ) <EOL> def __add__ ( self , other ) : <EOL> if not isinstance ( other , FileRestoreResponse ) : <EOL> raise NotImplemented ( ) <EOL> response = FileRestoreResponse ( self . username , self . items ) <EOL> response . num_items = self . num_items + other . num_items <EOL> self . response_body . seek ( <NUM_LIT:0> ) <EOL> other . response_body . seek ( <NUM_LIT:0> ) <EOL> shutil . copyfileobj ( self . response_body , response . response_body ) <EOL> shutil . copyfileobj ( other . response_body , response . response_body ) <EOL> return response <EOL> def finalize ( self ) : <EOL> """<STR_LIT>""" <EOL> with open ( self . get_filename ( ) , '<STR_LIT:w>' ) as response : <EOL> items = self . items_template . format ( self . num_items + <NUM_LIT:1> ) if self . items else '<STR_LIT>' <EOL> response . write ( self . start_tag_template . format ( <EOL> items = items , <EOL> username = self . username , <EOL> nature = ResponseNature . OTA_RESTORE_SUCCESS <EOL> ) ) <EOL> self . response_body . seek ( <NUM_LIT:0> ) <EOL> shutil . copyfileobj ( self . response_body , response ) <EOL> response . write ( self . closing_tag ) <EOL> self . finalized = True <EOL> self . close ( ) <EOL> def get_cache_payload ( self , full = False ) : <EOL> return { <EOL> '<STR_LIT:data>' : self . get_filename ( ) if not full else open ( self . get_filename ( ) , '<STR_LIT:r>' ) <EOL> } <EOL> def as_string ( self ) : <EOL> with open ( self . get_filename ( ) , '<STR_LIT:r>' ) as f : <EOL> return f . read ( ) <EOL> def get_http_response ( self ) : <EOL> headers = { '<STR_LIT>' : os . path . getsize ( self . get_filename ( ) ) } <EOL> return stream_response ( open ( self . get_filename ( ) , '<STR_LIT:r>' ) , headers ) <EOL> class CachedPayload ( object ) : <EOL> def __init__ ( self , payload , is_initial ) : <EOL> self . is_initial = is_initial <EOL> self . payload = payload <EOL> self . payload_path = None <EOL> if isinstance ( payload , dict ) : <EOL> self . payload_path = payload [ '<STR_LIT:data>' ] <EOL> if os . path . exists ( self . payload_path ) : <EOL> self . payload = open ( self . payload_path , '<STR_LIT:r>' ) <EOL> else : <EOL> self . payload = None <EOL> elif payload : <EOL> assert hasattr ( payload , '<STR_LIT>' ) , '<STR_LIT>' <EOL> def __nonzero__ ( self ) : <EOL> return bool ( self . payload ) <EOL> def as_string ( self ) : <EOL> try : <EOL> return self . payload . read ( ) <EOL> finally : <EOL> self . payload . close ( ) <EOL> def get_content_length ( self ) : <EOL> if self . payload_path : <EOL> return os . path . getsize ( self . payload_path ) <EOL> return None <EOL> def as_file ( self ) : <EOL> return self . payload <EOL> def finalize ( self ) : <EOL> if self and self . is_initial : <EOL> try : <EOL> file_reference = copy_payload_and_synclog_and_get_new_file ( self . payload ) <EOL> self . payload = file_reference . file <EOL> self . payload_path = file_reference . path <EOL> except Exception , e : <EOL> soft_assert ( to = [ '<STR_LIT>' + '<STR_LIT:@>' + '<STR_LIT>' ] ) ( False , u'<STR_LIT>' . format ( e ) ) <EOL> class CachedResponse ( object ) : <EOL> def __init__ ( self , payload ) : <EOL> self . payload = payload <EOL> def exists ( self ) : <EOL> return bool ( self . payload ) <EOL> def as_string ( self ) : <EOL> return self . payload . as_string ( ) <EOL> def get_http_response ( self ) : <EOL> headers = { } <EOL> content_length = self . payload . get_content_length ( ) <EOL> if content_length is not None : <EOL> headers [ '<STR_LIT>' ] = content_length <EOL> return stream_response ( self . payload . as_file ( ) , headers ) <EOL> class RestoreParams ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , sync_log_id = '<STR_LIT>' , version = V1 , state_hash = '<STR_LIT>' , include_item_count = False , <EOL> force_restore_mode = None , app = None ) : <EOL> self . sync_log_id = sync_log_id <EOL> self . version = version <EOL> self . state_hash = state_hash <EOL> self . include_item_count = include_item_count <EOL> self . force_restore_mode = force_restore_mode <EOL> self . app = app <EOL> class RestoreCacheSettings ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , force_cache = False , cache_timeout = None , overwrite_cache = False ) : <EOL> self . force_cache = force_cache <EOL> self . cache_timeout = cache_timeout if cache_timeout is not None else INITIAL_SYNC_CACHE_TIMEOUT <EOL> self . overwrite_cache = overwrite_cache <EOL> class RestoreState ( object ) : <EOL> """<STR_LIT>""" <EOL> restore_class = FileRestoreResponse <EOL> def __init__ ( self , project , user , params ) : <EOL> if not project or not project . name : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> self . project = project <EOL> self . domain = project . name <EOL> self . user = user <EOL> self . params = params <EOL> self . provider_log = { } <EOL> self . start_time = None <EOL> self . duration = None <EOL> self . current_sync_log = None <EOL> def validate_state ( self ) : <EOL> check_version ( self . params . version ) <EOL> if self . last_sync_log : <EOL> if self . params . state_hash : <EOL> parsed_hash = CaseStateHash . parse ( self . params . state_hash ) <EOL> computed_hash = self . last_sync_log . get_state_hash ( ) <EOL> if computed_hash != parsed_hash : <EOL> self . last_sync_log . had_state_error = True <EOL> self . last_sync_log . error_date = datetime . utcnow ( ) <EOL> self . last_sync_log . error_hash = str ( parsed_hash ) <EOL> self . last_sync_log . save ( ) <EOL> raise BadStateException ( <EOL> server_hash = computed_hash , <EOL> phone_hash = parsed_hash , <EOL> case_ids = self . last_sync_log . get_footprint_of_cases_on_phone ( ) <EOL> ) <EOL> @ property <EOL> @ memoized <EOL> def last_sync_log ( self ) : <EOL> if self . params . sync_log_id : <EOL> try : <EOL> sync_log = get_properly_wrapped_sync_log ( self . params . sync_log_id ) <EOL> except ResourceNotFound : <EOL> raise MissingSyncLog ( '<STR_LIT>' . format ( self . params . sync_log_id ) ) <EOL> if sync_log . doc_type != '<STR_LIT>' : <EOL> raise InvalidSyncLogException ( '<STR_LIT>' . format ( self . params . sync_log_id ) ) <EOL> elif sync_log . user_id != self . user . user_id : <EOL> raise SyncLogUserMismatch ( '<STR_LIT>' . format ( <EOL> self . params . sync_log_id , self . user . user_id , sync_log . user_id <EOL> ) ) <EOL> if not isinstance ( sync_log , self . sync_log_class ) : <EOL> sync_log = self . sync_log_class . from_other_format ( sync_log ) <EOL> return sync_log <EOL> else : <EOL> return None <EOL> @ property <EOL> def is_initial ( self ) : <EOL> return self . last_sync_log is None <EOL> @ property <EOL> def version ( self ) : <EOL> return self . params . version <EOL> @ property <EOL> @ memoized <EOL> def owner_ids ( self ) : <EOL> return set ( self . user . get_owner_ids ( ) ) <EOL> @ property <EOL> @ memoized <EOL> def stock_settings ( self ) : <EOL> if self . project and self . project . commtrack_settings : <EOL> return self . project . commtrack_settings . get_ota_restore_settings ( ) <EOL> else : <EOL> return StockSettings ( ) <EOL> @ property <EOL> def is_first_extension_sync ( self ) : <EOL> extension_toggle_enabled = EXTENSION_CASES_SYNC_ENABLED . enabled ( self . domain ) <EOL> try : <EOL> return extension_toggle_enabled and not self . last_sync_log . extensions_checked <EOL> except AttributeError : <EOL> return extension_toggle_enabled <EOL> def start_sync ( self ) : <EOL> self . start_time = datetime . utcnow ( ) <EOL> self . current_sync_log = self . create_sync_log ( ) <EOL> def finish_sync ( self ) : <EOL> self . duration = datetime . utcnow ( ) - self . start_time <EOL> self . current_sync_log . duration = self . duration . seconds <EOL> self . current_sync_log . save ( ) <EOL> def create_sync_log ( self ) : <EOL> previous_log_id = None if self . is_initial else self . last_sync_log . _id <EOL> previous_log_rev = None if self . is_initial else self . last_sync_log . _rev <EOL> last_seq = str ( get_db ( ) . info ( ) [ "<STR_LIT>" ] ) <EOL> new_synclog = SyncLog ( <EOL> user_id = self . user . user_id , <EOL> last_seq = last_seq , <EOL> owner_ids_on_phone = list ( self . owner_ids ) , <EOL> date = datetime . utcnow ( ) , <EOL> previous_log_id = previous_log_id , <EOL> previous_log_rev = previous_log_rev , <EOL> ) <EOL> new_synclog . save ( ** get_safe_write_kwargs ( ) ) <EOL> return new_synclog <EOL> @ property <EOL> def sync_log_class ( self ) : <EOL> return get_sync_log_class_by_format ( LOG_FORMAT_SIMPLIFIED ) <EOL> @ property <EOL> @ memoized <EOL> def loadtest_factor ( self ) : <EOL> return get_loadtest_factor ( self . domain , self . user ) <EOL> class RestoreConfig ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , project = None , user = None , params = None , cache_settings = None ) : <EOL> self . project = project <EOL> self . domain = project . name if project else '<STR_LIT>' <EOL> self . user = user <EOL> self . params = params or RestoreParams ( ) <EOL> self . cache_settings = cache_settings or RestoreCacheSettings ( ) <EOL> self . version = self . params . version <EOL> self . restore_state = RestoreState ( self . project , self . user , self . params ) <EOL> self . force_cache = self . cache_settings . force_cache <EOL> self . cache_timeout = self . cache_settings . cache_timeout <EOL> self . overwrite_cache = self . cache_settings . overwrite_cache <EOL> self . cache = get_redis_default_cache ( ) <EOL> @ property <EOL> @ memoized <EOL> def sync_log ( self ) : <EOL> return self . restore_state . last_sync_log <EOL> def validate ( self ) : <EOL> try : <EOL> self . restore_state . validate_state ( ) <EOL> except InvalidSyncLogException , e : <EOL> if LOOSE_SYNC_TOKEN_VALIDATION . enabled ( self . domain ) : <EOL> raise RestoreException ( e ) <EOL> else : <EOL> raise <EOL> def get_payload ( self ) : <EOL> """<STR_LIT>""" <EOL> self . validate ( ) <EOL> cached_response = self . get_cached_payload ( ) <EOL> if cached_response . exists ( ) : <EOL> return cached_response <EOL> self . restore_state . start_sync ( ) <EOL> with self . restore_state . restore_class ( <EOL> self . user . username , items = self . params . include_item_count ) as response : <EOL> normal_providers = get_restore_providers ( ) <EOL> for provider in normal_providers : <EOL> for element in provider . get_elements ( self . restore_state ) : <EOL> if element . tag == '<STR_LIT>' and len ( element ) == <NUM_LIT:0> : <EOL> ElementTree . SubElement ( element , '<STR_LIT>' ) <EOL> response . append ( element ) <EOL> long_running_providers = get_long_running_providers ( ) <EOL> for provider in long_running_providers : <EOL> partial_response = provider . get_response ( self . restore_state ) <EOL> response = response + partial_response <EOL> partial_response . close ( ) <EOL> response . finalize ( ) <EOL> self . restore_state . finish_sync ( ) <EOL> self . set_cached_payload_if_necessary ( response , self . restore_state . duration ) <EOL> return response <EOL> def get_response ( self ) : <EOL> try : <EOL> payload = self . get_payload ( ) <EOL> return payload . get_http_response ( ) <EOL> except RestoreException , e : <EOL> logging . exception ( "<STR_LIT>" % <EOL> ( type ( e ) . __name__ , self . user . username , str ( e ) ) ) <EOL> response = get_simple_response_xml ( <EOL> e . message , <EOL> ResponseNature . OTA_RESTORE_ERROR <EOL> ) <EOL> return HttpResponse ( response , content_type = "<STR_LIT>" , <EOL> status = <NUM_LIT> ) <EOL> def _initial_cache_key ( self ) : <EOL> return hashlib . md5 ( '<STR_LIT>' . format ( <EOL> user = self . user . user_id , <EOL> version = self . version , <EOL> ) ) . hexdigest ( ) <EOL> def get_cached_payload ( self ) : <EOL> if self . overwrite_cache : <EOL> return CachedResponse ( None ) <EOL> if self . sync_log : <EOL> payload = CachedPayload ( self . sync_log . get_cached_payload ( self . version , stream = True ) , is_initial = False ) <EOL> else : <EOL> payload = CachedPayload ( self . cache . get ( self . _initial_cache_key ( ) ) , is_initial = True ) <EOL> payload . finalize ( ) <EOL> return CachedResponse ( payload ) <EOL> def set_cached_payload_if_necessary ( self , resp , duration ) : <EOL> cache_payload = resp . get_cache_payload ( bool ( self . sync_log ) ) <EOL> if self . sync_log : <EOL> try : <EOL> data = cache_payload [ '<STR_LIT:data>' ] <EOL> self . sync_log . last_cached = datetime . utcnow ( ) <EOL> self . sync_log . hash_at_last_cached = str ( self . sync_log . get_state_hash ( ) ) <EOL> self . sync_log . save ( ) <EOL> self . sync_log . set_cached_payload ( data , self . version ) <EOL> try : <EOL> data . close ( ) <EOL> except AttributeError : <EOL> pass <EOL> except ResourceConflict : <EOL> pass <EOL> else : <EOL> if self . force_cache or duration > timedelta ( seconds = INITIAL_SYNC_CACHE_THRESHOLD ) : <EOL> self . cache . set ( self . _initial_cache_key ( ) , cache_payload , self . cache_timeout ) </s>
