<s> """<STR_LIT>""" <EOL> import csv <EOL> import errno <EOL> import getopt <EOL> import getpass <EOL> import imp <EOL> import logging <EOL> import os <EOL> import Queue <EOL> import re <EOL> import shutil <EOL> import signal <EOL> import StringIO <EOL> import sys <EOL> import threading <EOL> import time <EOL> import traceback <EOL> import urllib2 <EOL> import urlparse <EOL> from google . appengine . datastore import entity_pb <EOL> from google . appengine . api import apiproxy_stub_map <EOL> from google . appengine . api import datastore <EOL> from google . appengine . api import datastore_errors <EOL> from google . appengine . datastore import datastore_pb <EOL> from google . appengine . ext import db <EOL> from google . appengine . ext import key_range as key_range_module <EOL> from google . appengine . ext . db import polymodel <EOL> from google . appengine . ext . remote_api import remote_api_stub <EOL> from google . appengine . ext . remote_api import throttle as remote_api_throttle <EOL> from google . appengine . runtime import apiproxy_errors <EOL> from google . appengine . tools import adaptive_thread_pool <EOL> from google . appengine . tools import appengine_rpc <EOL> from google . appengine . tools . requeue import ReQueue <EOL> try : <EOL> import sqlite3 <EOL> except ImportError : <EOL> pass <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> KeyRange = key_range_module . KeyRange <EOL> DEFAULT_THREAD_COUNT = <NUM_LIT:10> <EOL> DEFAULT_BATCH_SIZE = <NUM_LIT:10> <EOL> DEFAULT_DOWNLOAD_BATCH_SIZE = <NUM_LIT:100> <EOL> DEFAULT_QUEUE_SIZE = DEFAULT_THREAD_COUNT * <NUM_LIT:10> <EOL> _THREAD_SHOULD_EXIT = '<STR_LIT>' <EOL> STATE_READ = <NUM_LIT:0> <EOL> STATE_SENDING = <NUM_LIT:1> <EOL> STATE_SENT = <NUM_LIT:2> <EOL> STATE_NOT_SENT = <NUM_LIT:3> <EOL> STATE_GETTING = <NUM_LIT:1> <EOL> STATE_GOT = <NUM_LIT:2> <EOL> STATE_ERROR = <NUM_LIT:3> <EOL> DATA_CONSUMED_TO_HERE = '<STR_LIT>' <EOL> INITIAL_BACKOFF = <NUM_LIT:1.0> <EOL> BACKOFF_FACTOR = <NUM_LIT> <EOL> DEFAULT_BANDWIDTH_LIMIT = <NUM_LIT> <EOL> DEFAULT_RPS_LIMIT = <NUM_LIT:20> <EOL> DEFAULT_REQUEST_LIMIT = <NUM_LIT:8> <EOL> MAXIMUM_INCREASE_DURATION = <NUM_LIT> <EOL> MAXIMUM_HOLD_DURATION = <NUM_LIT> <EOL> def ImportStateMessage ( state ) : <EOL> """<STR_LIT>""" <EOL> return ( { <EOL> STATE_READ : '<STR_LIT>' , <EOL> STATE_SENDING : '<STR_LIT>' , <EOL> STATE_SENT : '<STR_LIT>' , <EOL> STATE_NOT_SENT : '<STR_LIT>' <EOL> } [ state ] ) <EOL> def ExportStateMessage ( state ) : <EOL> """<STR_LIT>""" <EOL> return ( { <EOL> STATE_READ : '<STR_LIT>' , <EOL> STATE_GETTING : '<STR_LIT>' , <EOL> STATE_GOT : '<STR_LIT>' , <EOL> STATE_ERROR : '<STR_LIT>' <EOL> } [ state ] ) <EOL> def MapStateMessage ( state ) : <EOL> """<STR_LIT>""" <EOL> return ( { <EOL> STATE_READ : '<STR_LIT>' , <EOL> STATE_GETTING : '<STR_LIT>' , <EOL> STATE_GOT : '<STR_LIT>' , <EOL> STATE_ERROR : '<STR_LIT>' <EOL> } [ state ] ) <EOL> def ExportStateName ( state ) : <EOL> """<STR_LIT>""" <EOL> return ( { <EOL> STATE_READ : '<STR_LIT>' , <EOL> STATE_GETTING : '<STR_LIT>' , <EOL> STATE_GOT : '<STR_LIT>' , <EOL> STATE_ERROR : '<STR_LIT>' <EOL> } [ state ] ) <EOL> def ImportStateName ( state ) : <EOL> """<STR_LIT>""" <EOL> return ( { <EOL> STATE_READ : '<STR_LIT>' , <EOL> STATE_GETTING : '<STR_LIT>' , <EOL> STATE_GOT : '<STR_LIT>' , <EOL> STATE_NOT_SENT : '<STR_LIT>' <EOL> } [ state ] ) <EOL> class Error ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class MissingPropertyError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class FatalServerError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class ResumeError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class ConfigurationError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class AuthenticationError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class FileNotFoundError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class FileNotReadableError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class FileExistsError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class FileNotWritableError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class BadStateError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class KeyRangeError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class FieldSizeLimitError ( Error ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , limit ) : <EOL> self . message = """<STR_LIT>""" % limit <EOL> Error . __init__ ( self , self . message ) <EOL> class NameClashError ( Error ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , old_name , new_name , klass ) : <EOL> Error . __init__ ( self , old_name , new_name , klass ) <EOL> self . old_name = old_name <EOL> self . new_name = new_name <EOL> self . klass = klass <EOL> def GetCSVGeneratorFactory ( kind , csv_filename , batch_size , csv_has_header , <EOL> openfile = open , create_csv_reader = csv . reader ) : <EOL> """<STR_LIT>""" <EOL> loader = Loader . RegisteredLoader ( kind ) <EOL> loader . _Loader__openfile = openfile <EOL> loader . _Loader__create_csv_reader = create_csv_reader <EOL> record_generator = loader . generate_records ( csv_filename ) <EOL> def CreateGenerator ( request_manager , progress_queue , progress_generator ) : <EOL> """<STR_LIT>""" <EOL> return UploadWorkItemGenerator ( request_manager , <EOL> progress_queue , <EOL> progress_generator , <EOL> record_generator , <EOL> csv_has_header , <EOL> batch_size ) <EOL> return CreateGenerator <EOL> class UploadWorkItemGenerator ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> request_manager , <EOL> progress_queue , <EOL> progress_generator , <EOL> record_generator , <EOL> skip_first , <EOL> batch_size ) : <EOL> """<STR_LIT>""" <EOL> self . request_manager = request_manager <EOL> self . progress_queue = progress_queue <EOL> self . progress_generator = progress_generator <EOL> self . reader = record_generator <EOL> self . skip_first = skip_first <EOL> self . batch_size = batch_size <EOL> self . line_number = <NUM_LIT:1> <EOL> self . column_count = None <EOL> self . read_rows = [ ] <EOL> self . row_count = <NUM_LIT:0> <EOL> self . xfer_count = <NUM_LIT:0> <EOL> def _AdvanceTo ( self , line ) : <EOL> """<STR_LIT>""" <EOL> while self . line_number < line : <EOL> self . reader . next ( ) <EOL> self . line_number += <NUM_LIT:1> <EOL> self . row_count += <NUM_LIT:1> <EOL> self . xfer_count += <NUM_LIT:1> <EOL> def _ReadRows ( self , key_start , key_end ) : <EOL> """<STR_LIT>""" <EOL> assert self . line_number == key_start <EOL> self . read_rows = [ ] <EOL> while self . line_number <= key_end : <EOL> row = self . reader . next ( ) <EOL> self . row_count += <NUM_LIT:1> <EOL> if self . column_count is None : <EOL> self . column_count = len ( row ) <EOL> else : <EOL> if self . column_count != len ( row ) : <EOL> raise ResumeError ( '<STR_LIT>' % <EOL> ( self . column_count , str ( row ) ) ) <EOL> self . read_rows . append ( ( self . line_number , row ) ) <EOL> self . line_number += <NUM_LIT:1> <EOL> def _MakeItem ( self , key_start , key_end , rows , progress_key = None ) : <EOL> """<STR_LIT>""" <EOL> assert rows <EOL> item = UploadWorkItem ( self . request_manager , self . progress_queue , rows , <EOL> key_start , key_end , progress_key = progress_key ) <EOL> return item <EOL> def Batches ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . skip_first : <EOL> logger . info ( '<STR_LIT>' ) <EOL> try : <EOL> self . reader . next ( ) <EOL> except StopIteration : <EOL> return <EOL> exhausted = False <EOL> self . line_number = <NUM_LIT:1> <EOL> self . column_count = None <EOL> logger . info ( '<STR_LIT>' , <EOL> self . batch_size ) <EOL> state = None <EOL> if self . progress_generator : <EOL> for progress_key , state , key_start , key_end in self . progress_generator : <EOL> if key_start : <EOL> try : <EOL> self . _AdvanceTo ( key_start ) <EOL> self . _ReadRows ( key_start , key_end ) <EOL> yield self . _MakeItem ( key_start , <EOL> key_end , <EOL> self . read_rows , <EOL> progress_key = progress_key ) <EOL> except StopIteration : <EOL> logger . error ( '<STR_LIT>' ) <EOL> raise ResumeError ( <EOL> '<STR_LIT>' ) <EOL> elif state == DATA_CONSUMED_TO_HERE : <EOL> try : <EOL> self . _AdvanceTo ( key_end + <NUM_LIT:1> ) <EOL> except StopIteration : <EOL> state = None <EOL> if self . progress_generator is None or state == DATA_CONSUMED_TO_HERE : <EOL> while not exhausted : <EOL> key_start = self . line_number <EOL> key_end = self . line_number + self . batch_size - <NUM_LIT:1> <EOL> try : <EOL> self . _ReadRows ( key_start , key_end ) <EOL> except StopIteration : <EOL> exhausted = True <EOL> key_end = self . line_number - <NUM_LIT:1> <EOL> if key_start <= key_end : <EOL> yield self . _MakeItem ( key_start , key_end , self . read_rows ) <EOL> class CSVGenerator ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> csv_filename , <EOL> openfile = open , <EOL> create_csv_reader = csv . reader ) : <EOL> """<STR_LIT>""" <EOL> self . csv_filename = csv_filename <EOL> self . openfile = openfile <EOL> self . create_csv_reader = create_csv_reader <EOL> def Records ( self ) : <EOL> """<STR_LIT>""" <EOL> csv_file = self . openfile ( self . csv_filename , '<STR_LIT:rb>' ) <EOL> reader = self . create_csv_reader ( csv_file , skipinitialspace = True ) <EOL> try : <EOL> for record in reader : <EOL> yield record <EOL> except csv . Error , e : <EOL> if e . args and e . args [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) : <EOL> limit = e . args [ <NUM_LIT:1> ] <EOL> raise FieldSizeLimitError ( limit ) <EOL> else : <EOL> raise <EOL> class KeyRangeItemGenerator ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , request_manager , kind , progress_queue , progress_generator , <EOL> key_range_item_factory ) : <EOL> """<STR_LIT>""" <EOL> self . request_manager = request_manager <EOL> self . kind = kind <EOL> self . row_count = <NUM_LIT:0> <EOL> self . xfer_count = <NUM_LIT:0> <EOL> self . progress_queue = progress_queue <EOL> self . progress_generator = progress_generator <EOL> self . key_range_item_factory = key_range_item_factory <EOL> def Batches ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . progress_generator is not None : <EOL> for progress_key , state , key_start , key_end in self . progress_generator : <EOL> if state is not None and state != STATE_GOT and key_start is not None : <EOL> key_start = ParseKey ( key_start ) <EOL> key_end = ParseKey ( key_end ) <EOL> key_range = KeyRange ( key_start = key_start , <EOL> key_end = key_end ) <EOL> result = self . key_range_item_factory ( self . request_manager , <EOL> self . progress_queue , <EOL> self . kind , <EOL> key_range , <EOL> progress_key = progress_key , <EOL> state = STATE_READ ) <EOL> yield result <EOL> else : <EOL> key_range = KeyRange ( ) <EOL> yield self . key_range_item_factory ( self . request_manager , <EOL> self . progress_queue , <EOL> self . kind , <EOL> key_range ) <EOL> class DownloadResult ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , continued , direction , keys , entities ) : <EOL> self . continued = continued <EOL> self . direction = direction <EOL> self . keys = keys <EOL> self . entities = entities <EOL> self . count = len ( keys ) <EOL> assert self . count == len ( entities ) <EOL> assert direction in ( key_range_module . KeyRange . ASC , <EOL> key_range_module . KeyRange . DESC ) <EOL> if self . count > <NUM_LIT:0> : <EOL> if direction == key_range_module . KeyRange . ASC : <EOL> self . key_start = keys [ <NUM_LIT:0> ] <EOL> self . key_end = keys [ - <NUM_LIT:1> ] <EOL> else : <EOL> self . key_start = keys [ - <NUM_LIT:1> ] <EOL> self . key_end = keys [ <NUM_LIT:0> ] <EOL> def Entities ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . direction == key_range_module . KeyRange . ASC : <EOL> return list ( self . entities ) <EOL> else : <EOL> result = list ( self . entities ) <EOL> result . reverse ( ) <EOL> return result <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % ( <EOL> str ( self . continued ) , '<STR_LIT:\n>' . join ( str ( self . entities ) ) ) <EOL> class _WorkItem ( adaptive_thread_pool . WorkItem ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , progress_queue , key_start , key_end , state_namer , <EOL> state = STATE_READ , progress_key = None ) : <EOL> """<STR_LIT>""" <EOL> adaptive_thread_pool . WorkItem . __init__ ( self , <EOL> '<STR_LIT>' % ( key_start , key_end ) ) <EOL> self . progress_queue = progress_queue <EOL> self . state_namer = state_namer <EOL> self . state = state <EOL> self . progress_key = progress_key <EOL> self . progress_event = threading . Event ( ) <EOL> self . key_start = key_start <EOL> self . key_end = key_end <EOL> self . error = None <EOL> self . traceback = None <EOL> def _TransferItem ( self , thread_pool ) : <EOL> raise NotImplementedError ( ) <EOL> def SetError ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . error : <EOL> exc_info = sys . exc_info ( ) <EOL> self . error = exc_info [ <NUM_LIT:1> ] <EOL> self . traceback = exc_info [ <NUM_LIT:2> ] <EOL> def PerformWork ( self , thread_pool ) : <EOL> """<STR_LIT>""" <EOL> status = adaptive_thread_pool . WorkItem . FAILURE <EOL> instruction = adaptive_thread_pool . ThreadGate . DECREASE <EOL> try : <EOL> self . MarkAsTransferring ( ) <EOL> try : <EOL> transfer_time = self . _TransferItem ( thread_pool ) <EOL> if transfer_time is None : <EOL> status = adaptive_thread_pool . WorkItem . RETRY <EOL> instruction = adaptive_thread_pool . ThreadGate . HOLD <EOL> else : <EOL> logger . debug ( '<STR_LIT>' , <EOL> threading . currentThread ( ) . getName ( ) , self , self . count , <EOL> transfer_time ) <EOL> sys . stdout . write ( '<STR_LIT:.>' ) <EOL> sys . stdout . flush ( ) <EOL> status = adaptive_thread_pool . WorkItem . SUCCESS <EOL> if transfer_time <= MAXIMUM_INCREASE_DURATION : <EOL> instruction = adaptive_thread_pool . ThreadGate . INCREASE <EOL> elif transfer_time <= MAXIMUM_HOLD_DURATION : <EOL> instruction = adaptive_thread_pool . ThreadGate . HOLD <EOL> except ( db . InternalError , db . NotSavedError , db . Timeout , <EOL> db . TransactionFailedError , <EOL> apiproxy_errors . OverQuotaError , <EOL> apiproxy_errors . DeadlineExceededError , <EOL> apiproxy_errors . ApplicationError ) , e : <EOL> status = adaptive_thread_pool . WorkItem . RETRY <EOL> logger . exception ( '<STR_LIT>' , e ) <EOL> except urllib2 . HTTPError , e : <EOL> http_status = e . code <EOL> if http_status == <NUM_LIT> or ( http_status >= <NUM_LIT> and http_status < <NUM_LIT> ) : <EOL> status = adaptive_thread_pool . WorkItem . RETRY <EOL> logger . exception ( '<STR_LIT>' , <EOL> http_status , e . msg ) <EOL> else : <EOL> self . SetError ( ) <EOL> status = adaptive_thread_pool . WorkItem . FAILURE <EOL> except urllib2 . URLError , e : <EOL> if IsURLErrorFatal ( e ) : <EOL> self . SetError ( ) <EOL> status = adaptive_thread_pool . WorkItem . FAILURE <EOL> else : <EOL> status = adaptive_thread_pool . WorkItem . RETRY <EOL> logger . exception ( '<STR_LIT>' , e . reason ) <EOL> finally : <EOL> if status == adaptive_thread_pool . WorkItem . SUCCESS : <EOL> self . MarkAsTransferred ( ) <EOL> else : <EOL> self . MarkAsError ( ) <EOL> return ( status , instruction ) <EOL> def _AssertInState ( self , * states ) : <EOL> """<STR_LIT>""" <EOL> if not self . state in states : <EOL> raise BadStateError ( '<STR_LIT>' % <EOL> ( str ( self ) , <EOL> self . state_namer ( self . state ) , <EOL> map ( self . state_namer , states ) ) ) <EOL> def _AssertProgressKey ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . progress_key is None : <EOL> raise BadStateError ( '<STR_LIT>' % str ( self ) ) <EOL> def MarkAsRead ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _AssertInState ( STATE_READ ) <EOL> self . _StateTransition ( STATE_READ , blocking = True ) <EOL> def MarkAsTransferring ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _AssertInState ( STATE_READ , STATE_ERROR ) <EOL> self . _AssertProgressKey ( ) <EOL> self . _StateTransition ( STATE_GETTING , blocking = True ) <EOL> def MarkAsTransferred ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def MarkAsError ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _AssertInState ( STATE_GETTING ) <EOL> self . _AssertProgressKey ( ) <EOL> self . _StateTransition ( STATE_ERROR , blocking = True ) <EOL> def _StateTransition ( self , new_state , blocking = False ) : <EOL> """<STR_LIT>""" <EOL> assert not self . progress_event . isSet ( ) <EOL> self . state = new_state <EOL> self . progress_queue . put ( self ) <EOL> if blocking : <EOL> self . progress_event . wait ( ) <EOL> self . progress_event . clear ( ) <EOL> class UploadWorkItem ( _WorkItem ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , request_manager , progress_queue , rows , key_start , key_end , <EOL> progress_key = None ) : <EOL> """<STR_LIT>""" <EOL> _WorkItem . __init__ ( self , progress_queue , key_start , key_end , <EOL> ImportStateName , state = STATE_READ , <EOL> progress_key = progress_key ) <EOL> assert isinstance ( key_start , ( int , long ) ) <EOL> assert isinstance ( key_end , ( int , long ) ) <EOL> assert key_start <= key_end <EOL> self . request_manager = request_manager <EOL> self . rows = rows <EOL> self . content = None <EOL> self . count = len ( rows ) <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % ( self . key_start , self . key_end ) <EOL> def _TransferItem ( self , thread_pool , get_time = time . time ) : <EOL> """<STR_LIT>""" <EOL> t = get_time ( ) <EOL> if not self . content : <EOL> self . content = self . request_manager . EncodeContent ( self . rows ) <EOL> try : <EOL> self . request_manager . PostEntities ( self . content ) <EOL> except : <EOL> raise <EOL> return get_time ( ) - t <EOL> def MarkAsTransferred ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _AssertInState ( STATE_SENDING ) <EOL> self . _AssertProgressKey ( ) <EOL> self . _StateTransition ( STATE_SENT , blocking = False ) <EOL> def GetImplementationClass ( kind_or_class_key ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( kind_or_class_key , tuple ) : <EOL> try : <EOL> implementation_class = polymodel . _class_map [ kind_or_class_key ] <EOL> except KeyError : <EOL> raise db . KindError ( '<STR_LIT>' % <EOL> kind_or_class_key ) <EOL> else : <EOL> implementation_class = db . class_for_kind ( kind_or_class_key ) <EOL> return implementation_class <EOL> def KeyLEQ ( key1 , key2 ) : <EOL> """<STR_LIT>""" <EOL> if key1 is None or key2 is None : <EOL> return True <EOL> return key1 <= key2 <EOL> class KeyRangeItem ( _WorkItem ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> request_manager , <EOL> progress_queue , <EOL> kind , <EOL> key_range , <EOL> progress_key = None , <EOL> state = STATE_READ ) : <EOL> """<STR_LIT>""" <EOL> _WorkItem . __init__ ( self , progress_queue , key_range . key_start , <EOL> key_range . key_end , ExportStateName , state = state , <EOL> progress_key = progress_key ) <EOL> self . request_manager = request_manager <EOL> self . kind = kind <EOL> self . key_range = key_range <EOL> self . download_result = None <EOL> self . count = <NUM_LIT:0> <EOL> self . key_start = key_range . key_start <EOL> self . key_end = key_range . key_end <EOL> def __str__ ( self ) : <EOL> return str ( self . key_range ) <EOL> def __repr__ ( self ) : <EOL> return self . __str__ ( ) <EOL> def MarkAsTransferred ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def Process ( self , download_result , thread_pool , batch_size , <EOL> new_state = STATE_GOT ) : <EOL> """<STR_LIT>""" <EOL> self . _AssertInState ( STATE_GETTING ) <EOL> self . _AssertProgressKey ( ) <EOL> self . download_result = download_result <EOL> self . count = len ( download_result . keys ) <EOL> if download_result . continued : <EOL> self . _FinishedRange ( ) . _StateTransition ( new_state , blocking = True ) <EOL> self . _AddUnfinishedRanges ( thread_pool , batch_size ) <EOL> else : <EOL> self . _StateTransition ( new_state , blocking = True ) <EOL> def _FinishedRange ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . download_result is not None <EOL> if self . key_range . direction == key_range_module . KeyRange . ASC : <EOL> key_start = self . key_range . key_start <EOL> if self . download_result . continued : <EOL> key_end = self . download_result . key_end <EOL> else : <EOL> key_end = self . key_range . key_end <EOL> else : <EOL> key_end = self . key_range . key_end <EOL> if self . download_result . continued : <EOL> key_start = self . download_result . key_start <EOL> else : <EOL> key_start = self . key_range . key_start <EOL> key_range = KeyRange ( key_start = key_start , <EOL> key_end = key_end , <EOL> direction = self . key_range . direction ) <EOL> result = self . __class__ ( self . request_manager , <EOL> self . progress_queue , <EOL> self . kind , <EOL> key_range , <EOL> progress_key = self . progress_key , <EOL> state = self . state ) <EOL> result . download_result = self . download_result <EOL> result . count = self . count <EOL> return result <EOL> def _SplitAndAddRanges ( self , thread_pool , batch_size ) : <EOL> """<STR_LIT>""" <EOL> if self . download_result . direction == key_range_module . KeyRange . ASC : <EOL> key_range = KeyRange ( <EOL> key_start = self . download_result . key_end , <EOL> key_end = self . key_range . key_end , <EOL> include_start = False ) <EOL> else : <EOL> key_range = KeyRange ( <EOL> key_start = self . key_range . key_start , <EOL> key_end = self . download_result . key_start , <EOL> include_end = False ) <EOL> if thread_pool . QueuedItemCount ( ) > <NUM_LIT:2> * thread_pool . num_threads ( ) : <EOL> ranges = [ key_range ] <EOL> else : <EOL> ranges = key_range . split_range ( batch_size = batch_size ) <EOL> for key_range in ranges : <EOL> key_range_item = self . __class__ ( self . request_manager , <EOL> self . progress_queue , <EOL> self . kind , <EOL> key_range ) <EOL> key_range_item . MarkAsRead ( ) <EOL> thread_pool . SubmitItem ( key_range_item , block = True ) <EOL> def _AddUnfinishedRanges ( self , thread_pool , batch_size ) : <EOL> """<STR_LIT>""" <EOL> assert self . download_result is not None <EOL> if self . download_result . continued : <EOL> self . _SplitAndAddRanges ( thread_pool , batch_size ) <EOL> else : <EOL> raise KeyRangeError ( '<STR_LIT>' ) <EOL> class DownloadItem ( KeyRangeItem ) : <EOL> """<STR_LIT>""" <EOL> def _TransferItem ( self , thread_pool , get_time = time . time ) : <EOL> """<STR_LIT>""" <EOL> t = get_time ( ) <EOL> download_result = self . request_manager . GetEntities ( self ) <EOL> transfer_time = get_time ( ) - t <EOL> self . Process ( download_result , thread_pool , <EOL> self . request_manager . batch_size ) <EOL> return transfer_time <EOL> class MapperItem ( KeyRangeItem ) : <EOL> """<STR_LIT>""" <EOL> def _TransferItem ( self , thread_pool , get_time = time . time ) : <EOL> t = get_time ( ) <EOL> download_result = self . request_manager . GetEntities ( self ) <EOL> transfer_time = get_time ( ) - t <EOL> mapper = self . request_manager . GetMapper ( ) <EOL> try : <EOL> mapper . batch_apply ( download_result . Entities ( ) ) <EOL> except MapperRetry : <EOL> return None <EOL> self . Process ( download_result , thread_pool , <EOL> self . request_manager . batch_size ) <EOL> return transfer_time <EOL> class RequestManager ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> app_id , <EOL> host_port , <EOL> url_path , <EOL> kind , <EOL> throttle , <EOL> batch_size , <EOL> secure , <EOL> email , <EOL> passin , <EOL> dry_run = False ) : <EOL> """<STR_LIT>""" <EOL> self . app_id = app_id <EOL> self . host_port = host_port <EOL> self . host = host_port . split ( '<STR_LIT::>' ) [ <NUM_LIT:0> ] <EOL> if url_path and url_path [ <NUM_LIT:0> ] != '<STR_LIT:/>' : <EOL> url_path = '<STR_LIT:/>' + url_path <EOL> self . url_path = url_path <EOL> self . kind = kind <EOL> self . throttle = throttle <EOL> self . batch_size = batch_size <EOL> self . secure = secure <EOL> self . authenticated = False <EOL> self . auth_called = False <EOL> self . parallel_download = True <EOL> self . email = email <EOL> self . passin = passin <EOL> self . mapper = None <EOL> self . dry_run = dry_run <EOL> if self . dry_run : <EOL> logger . info ( '<STR_LIT>' ) <EOL> return <EOL> logger . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( url_path , host_port ) ) <EOL> def CookieHttpRpcServer ( * args , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = True <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return appengine_rpc . HttpRpcServer ( * args , ** kwargs ) <EOL> remote_api_stub . ConfigureRemoteDatastore ( <EOL> app_id , <EOL> url_path , <EOL> self . AuthFunction , <EOL> servername = host_port , <EOL> rpc_server_factory = CookieHttpRpcServer , <EOL> secure = self . secure ) <EOL> remote_api_throttle . ThrottleRemoteDatastore ( self . throttle ) <EOL> logger . debug ( '<STR_LIT>' , os . environ [ '<STR_LIT>' ] ) <EOL> def Authenticate ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( '<STR_LIT>' , self . host_port , self . url_path ) <EOL> if self . dry_run : <EOL> self . authenticated = True <EOL> return <EOL> remote_api_stub . MaybeInvokeAuthentication ( ) <EOL> self . authenticated = True <EOL> def AuthFunction ( self , <EOL> raw_input_fn = raw_input , <EOL> password_input_fn = getpass . getpass ) : <EOL> """<STR_LIT>""" <EOL> if self . email : <EOL> email = self . email <EOL> else : <EOL> print '<STR_LIT>' % ( <EOL> self . host ) <EOL> email = raw_input_fn ( '<STR_LIT>' ) <EOL> if email : <EOL> password_prompt = '<STR_LIT>' % email <EOL> if self . passin : <EOL> password = raw_input_fn ( password_prompt ) <EOL> else : <EOL> password = password_input_fn ( password_prompt ) <EOL> else : <EOL> password = None <EOL> self . auth_called = True <EOL> return ( email , password ) <EOL> def EncodeContent ( self , rows , loader = None ) : <EOL> """<STR_LIT>""" <EOL> if not loader : <EOL> try : <EOL> loader = Loader . RegisteredLoader ( self . kind ) <EOL> except KeyError : <EOL> logger . error ( '<STR_LIT>' % self . kind ) <EOL> raise ConfigurationError ( '<STR_LIT>' % self . kind ) <EOL> entities = [ ] <EOL> for line_number , values in rows : <EOL> key = loader . generate_key ( line_number , values ) <EOL> if isinstance ( key , datastore . Key ) : <EOL> parent = key . parent ( ) <EOL> key = key . name ( ) <EOL> else : <EOL> parent = None <EOL> entity = loader . create_entity ( values , key_name = key , parent = parent ) <EOL> def ToEntity ( entity ) : <EOL> if isinstance ( entity , db . Model ) : <EOL> return entity . _populate_entity ( ) <EOL> else : <EOL> return entity <EOL> if isinstance ( entity , list ) : <EOL> entities . extend ( map ( ToEntity , entity ) ) <EOL> elif entity : <EOL> entities . append ( ToEntity ( entity ) ) <EOL> return entities <EOL> def PostEntities ( self , entities ) : <EOL> """<STR_LIT>""" <EOL> if self . dry_run : <EOL> return <EOL> datastore . Put ( entities ) <EOL> def _QueryForPbs ( self , query ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> query_pb = query . _ToPb ( limit = self . batch_size ) <EOL> result_pb = datastore_pb . QueryResult ( ) <EOL> apiproxy_stub_map . MakeSyncCall ( '<STR_LIT>' , '<STR_LIT>' , query_pb , <EOL> result_pb ) <EOL> next_pb = datastore_pb . NextRequest ( ) <EOL> next_pb . set_count ( self . batch_size ) <EOL> next_pb . mutable_cursor ( ) . CopyFrom ( result_pb . cursor ( ) ) <EOL> result_pb = datastore_pb . QueryResult ( ) <EOL> apiproxy_stub_map . MakeSyncCall ( '<STR_LIT>' , '<STR_LIT>' , next_pb , result_pb ) <EOL> return result_pb . result_list ( ) <EOL> except apiproxy_errors . ApplicationError , e : <EOL> raise datastore . _ToDatastoreError ( e ) <EOL> def GetEntities ( self , key_range_item , key_factory = datastore . Key ) : <EOL> """<STR_LIT>""" <EOL> keys = [ ] <EOL> entities = [ ] <EOL> if self . parallel_download : <EOL> query = key_range_item . key_range . make_directed_datastore_query ( self . kind ) <EOL> try : <EOL> results = self . _QueryForPbs ( query ) <EOL> except datastore_errors . NeedIndexError : <EOL> logger . info ( '<STR_LIT>' <EOL> '<STR_LIT>' , self . kind ) <EOL> self . parallel_download = False <EOL> if not self . parallel_download : <EOL> key_range_item . key_range . direction = key_range_module . KeyRange . ASC <EOL> query = key_range_item . key_range . make_ascending_datastore_query ( self . kind ) <EOL> results = self . _QueryForPbs ( query ) <EOL> size = len ( results ) <EOL> for entity in results : <EOL> key = key_factory ( ) <EOL> key . _Key__reference = entity . key ( ) <EOL> entities . append ( entity ) <EOL> keys . append ( key ) <EOL> continued = ( size == self . batch_size ) <EOL> key_range_item . count = size <EOL> return DownloadResult ( continued , key_range_item . key_range . direction , <EOL> keys , entities ) <EOL> def GetMapper ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . mapper : <EOL> try : <EOL> self . mapper = Mapper . RegisteredMapper ( self . kind ) <EOL> except KeyError : <EOL> logger . error ( '<STR_LIT>' % self . kind ) <EOL> raise ConfigurationError ( '<STR_LIT>' % self . kind ) <EOL> return self . mapper <EOL> def InterruptibleSleep ( sleep_time ) : <EOL> """<STR_LIT>""" <EOL> slept = <NUM_LIT:0.0> <EOL> epsilon = <NUM_LIT> <EOL> thread = threading . currentThread ( ) <EOL> while slept < sleep_time - epsilon : <EOL> remaining = sleep_time - slept <EOL> this_sleep_time = min ( remaining , <NUM_LIT:0.5> ) <EOL> time . sleep ( this_sleep_time ) <EOL> slept += this_sleep_time <EOL> if thread . exit_flag : <EOL> return <EOL> class _ThreadBase ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> threading . Thread . __init__ ( self ) <EOL> self . setDaemon ( True ) <EOL> self . exit_flag = False <EOL> self . error = None <EOL> self . traceback = None <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( '<STR_LIT>' , self . getName ( ) , self . __class__ . __name__ ) <EOL> try : <EOL> self . PerformWork ( ) <EOL> except : <EOL> self . SetError ( ) <EOL> logger . exception ( '<STR_LIT>' , self . getName ( ) , self . __class__ . __name__ ) <EOL> logger . debug ( '<STR_LIT>' , self . getName ( ) , self . __class__ . __name__ ) <EOL> def SetError ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . error : <EOL> exc_info = sys . exc_info ( ) <EOL> self . error = exc_info [ <NUM_LIT:1> ] <EOL> self . traceback = exc_info [ <NUM_LIT:2> ] <EOL> def PerformWork ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def CheckError ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . error : <EOL> logger . error ( '<STR_LIT>' , self . GetFriendlyName ( ) , self . error ) <EOL> if self . traceback : <EOL> logger . debug ( '<STR_LIT>' . join ( traceback . format_exception ( self . error . __class__ , <EOL> self . error , <EOL> self . traceback ) ) ) <EOL> def GetFriendlyName ( self ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> return self . NAME <EOL> return '<STR_LIT>' <EOL> non_fatal_error_codes = set ( [ errno . EAGAIN , <EOL> errno . ENETUNREACH , <EOL> errno . ENETRESET , <EOL> errno . ECONNRESET , <EOL> errno . ETIMEDOUT , <EOL> errno . EHOSTUNREACH ] ) <EOL> def IsURLErrorFatal ( error ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( error , urllib2 . URLError ) <EOL> if not hasattr ( error , '<STR_LIT>' ) : <EOL> return True <EOL> if not isinstance ( error . reason [ <NUM_LIT:0> ] , int ) : <EOL> return True <EOL> return error . reason [ <NUM_LIT:0> ] not in non_fatal_error_codes <EOL> class DataSourceThread ( _ThreadBase ) : <EOL> """<STR_LIT>""" <EOL> NAME = '<STR_LIT>' <EOL> def __init__ ( self , <EOL> request_manager , <EOL> thread_pool , <EOL> progress_queue , <EOL> workitem_generator_factory , <EOL> progress_generator_factory ) : <EOL> """<STR_LIT>""" <EOL> _ThreadBase . __init__ ( self ) <EOL> self . request_manager = request_manager <EOL> self . thread_pool = thread_pool <EOL> self . progress_queue = progress_queue <EOL> self . workitem_generator_factory = workitem_generator_factory <EOL> self . progress_generator_factory = progress_generator_factory <EOL> self . entity_count = <NUM_LIT:0> <EOL> def PerformWork ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . progress_generator_factory : <EOL> progress_gen = self . progress_generator_factory ( ) <EOL> else : <EOL> progress_gen = None <EOL> content_gen = self . workitem_generator_factory ( self . request_manager , <EOL> self . progress_queue , <EOL> progress_gen ) <EOL> self . xfer_count = <NUM_LIT:0> <EOL> self . read_count = <NUM_LIT:0> <EOL> self . read_all = False <EOL> for item in content_gen . Batches ( ) : <EOL> item . MarkAsRead ( ) <EOL> while not self . exit_flag : <EOL> try : <EOL> self . thread_pool . SubmitItem ( item , block = True , timeout = <NUM_LIT:1.0> ) <EOL> self . entity_count += item . count <EOL> break <EOL> except Queue . Full : <EOL> pass <EOL> if self . exit_flag : <EOL> break <EOL> if not self . exit_flag : <EOL> self . read_all = True <EOL> self . read_count = content_gen . row_count <EOL> self . xfer_count = content_gen . xfer_count <EOL> def _RunningInThread ( thread ) : <EOL> """<STR_LIT>""" <EOL> return threading . currentThread ( ) . getName ( ) == thread . getName ( ) <EOL> class _Database ( object ) : <EOL> """<STR_LIT>""" <EOL> SIGNATURE_TABLE_NAME = '<STR_LIT>' <EOL> def __init__ ( self , <EOL> db_filename , <EOL> create_table , <EOL> signature , <EOL> index = None , <EOL> commit_periodicity = <NUM_LIT:100> ) : <EOL> """<STR_LIT>""" <EOL> self . db_filename = db_filename <EOL> logger . info ( '<STR_LIT>' , db_filename ) <EOL> self . primary_conn = sqlite3 . connect ( db_filename , isolation_level = None ) <EOL> self . primary_thread = threading . currentThread ( ) <EOL> self . secondary_conn = None <EOL> self . secondary_thread = None <EOL> self . operation_count = <NUM_LIT:0> <EOL> self . commit_periodicity = commit_periodicity <EOL> try : <EOL> self . primary_conn . execute ( create_table ) <EOL> except sqlite3 . OperationalError , e : <EOL> if '<STR_LIT>' not in e . message : <EOL> raise <EOL> if index : <EOL> try : <EOL> self . primary_conn . execute ( index ) <EOL> except sqlite3 . OperationalError , e : <EOL> if '<STR_LIT>' not in e . message : <EOL> raise <EOL> self . existing_table = False <EOL> signature_cursor = self . primary_conn . cursor ( ) <EOL> create_signature = """<STR_LIT>""" % _Database . SIGNATURE_TABLE_NAME <EOL> try : <EOL> self . primary_conn . execute ( create_signature ) <EOL> self . primary_conn . cursor ( ) . execute ( <EOL> '<STR_LIT>' % _Database . SIGNATURE_TABLE_NAME , <EOL> ( signature , ) ) <EOL> except sqlite3 . OperationalError , e : <EOL> if '<STR_LIT>' not in e . message : <EOL> logger . exception ( '<STR_LIT>' ) <EOL> raise <EOL> else : <EOL> self . existing_table = True <EOL> signature_cursor . execute ( <EOL> '<STR_LIT>' % _Database . SIGNATURE_TABLE_NAME ) <EOL> ( result , ) = signature_cursor . fetchone ( ) <EOL> if result and result != signature : <EOL> logger . error ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> result , signature ) <EOL> raise ResumeError ( '<STR_LIT>' % ( <EOL> signature , result ) ) <EOL> def ThreadComplete ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . secondary_conn : <EOL> self . _MaybeCommit ( force_commit = True ) <EOL> def _MaybeCommit ( self , force_commit = False ) : <EOL> """<STR_LIT>""" <EOL> self . operation_count += <NUM_LIT:1> <EOL> if force_commit or ( self . operation_count % self . commit_periodicity ) == <NUM_LIT:0> : <EOL> self . secondary_conn . commit ( ) <EOL> def _OpenSecondaryConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . secondary_conn : <EOL> return <EOL> assert not _RunningInThread ( self . primary_thread ) <EOL> self . secondary_thread = threading . currentThread ( ) <EOL> self . secondary_conn = sqlite3 . connect ( self . db_filename ) <EOL> self . insert_cursor = self . secondary_conn . cursor ( ) <EOL> self . update_cursor = self . secondary_conn . cursor ( ) <EOL> zero_matcher = re . compile ( r'<STR_LIT:\x00>' ) <EOL> zero_one_matcher = re . compile ( r'<STR_LIT>' ) <EOL> def KeyStr ( key ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( key , datastore . Key ) <EOL> path = key . to_path ( ) <EOL> out_path = [ ] <EOL> for part in path : <EOL> if isinstance ( part , ( int , long ) ) : <EOL> part = '<STR_LIT>' % part <EOL> else : <EOL> part = '<STR_LIT>' % part <EOL> out_path . append ( zero_matcher . sub ( u'<STR_LIT>' , part ) ) <EOL> out_str = u'<STR_LIT>' . join ( out_path ) <EOL> return out_str <EOL> def StrKey ( key_str ) : <EOL> """<STR_LIT>""" <EOL> parts = key_str . split ( u'<STR_LIT>' ) <EOL> for i in xrange ( len ( parts ) ) : <EOL> if parts [ i ] [ <NUM_LIT:0> ] == '<STR_LIT::>' : <EOL> part = parts [ i ] [ <NUM_LIT:1> : ] <EOL> part = zero_one_matcher . sub ( u'<STR_LIT>' , part ) <EOL> parts [ i ] = part <EOL> else : <EOL> parts [ i ] = int ( parts [ i ] ) <EOL> return datastore . Key . from_path ( * parts ) <EOL> class ResultDatabase ( _Database ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , db_filename , signature , commit_periodicity = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> self . complete = False <EOL> create_table = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> _Database . __init__ ( self , <EOL> db_filename , <EOL> create_table , <EOL> signature , <EOL> commit_periodicity = commit_periodicity ) <EOL> if self . existing_table : <EOL> cursor = self . primary_conn . cursor ( ) <EOL> cursor . execute ( '<STR_LIT>' ) <EOL> self . existing_count = int ( cursor . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> else : <EOL> self . existing_count = <NUM_LIT:0> <EOL> self . count = self . existing_count <EOL> def _StoreEntity ( self , entity_id , entity ) : <EOL> """<STR_LIT>""" <EOL> assert _RunningInThread ( self . secondary_thread ) <EOL> assert isinstance ( entity_id , datastore . Key ) , ( <EOL> '<STR_LIT>' % entity_id . __class__ . __name__ ) <EOL> key_str = buffer ( KeyStr ( entity_id ) . encode ( '<STR_LIT:utf-8>' ) ) <EOL> self . insert_cursor . execute ( <EOL> '<STR_LIT>' , ( key_str , ) ) <EOL> already_present = self . insert_cursor . fetchone ( ) [ <NUM_LIT:0> ] <EOL> result = True <EOL> if already_present : <EOL> result = False <EOL> self . insert_cursor . execute ( '<STR_LIT>' , <EOL> ( key_str , ) ) <EOL> else : <EOL> self . count += <NUM_LIT:1> <EOL> value = entity . Encode ( ) <EOL> self . insert_cursor . execute ( <EOL> '<STR_LIT>' , <EOL> ( key_str , buffer ( value ) ) ) <EOL> return result <EOL> def StoreEntities ( self , keys , entities ) : <EOL> """<STR_LIT>""" <EOL> self . _OpenSecondaryConnection ( ) <EOL> t = time . time ( ) <EOL> count = <NUM_LIT:0> <EOL> for entity_id , entity in zip ( keys , <EOL> entities ) : <EOL> if self . _StoreEntity ( entity_id , entity ) : <EOL> count += <NUM_LIT:1> <EOL> logger . debug ( '<STR_LIT>' , <EOL> self . db_filename , <EOL> time . time ( ) - t ) <EOL> logger . debug ( '<STR_LIT>' , self . count ) <EOL> self . _MaybeCommit ( ) <EOL> return count <EOL> def ResultsComplete ( self ) : <EOL> """<STR_LIT>""" <EOL> self . complete = True <EOL> def AllEntities ( self ) : <EOL> """<STR_LIT>""" <EOL> conn = sqlite3 . connect ( self . db_filename , isolation_level = None ) <EOL> cursor = conn . cursor ( ) <EOL> cursor . execute ( <EOL> '<STR_LIT>' ) <EOL> for unused_entity_id , entity in cursor : <EOL> entity_proto = entity_pb . EntityProto ( contents = entity ) <EOL> yield datastore . Entity . _FromPb ( entity_proto ) <EOL> class _ProgressDatabase ( _Database ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> db_filename , <EOL> sql_type , <EOL> py_type , <EOL> signature , <EOL> commit_periodicity = <NUM_LIT:100> ) : <EOL> """<STR_LIT>""" <EOL> self . prior_key_end = None <EOL> create_table = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( sql_type , sql_type ) ) <EOL> self . py_type = py_type <EOL> index = '<STR_LIT>' <EOL> _Database . __init__ ( self , <EOL> db_filename , <EOL> create_table , <EOL> signature , <EOL> index = index , <EOL> commit_periodicity = commit_periodicity ) <EOL> def UseProgressData ( self ) : <EOL> """<STR_LIT>""" <EOL> assert _RunningInThread ( self . primary_thread ) <EOL> cursor = self . primary_conn . cursor ( ) <EOL> cursor . execute ( '<STR_LIT>' ) <EOL> row = cursor . fetchone ( ) <EOL> if row is None : <EOL> raise ResumeError ( '<STR_LIT>' ) <EOL> return row [ <NUM_LIT:0> ] != <NUM_LIT:0> <EOL> def StoreKeys ( self , key_start , key_end ) : <EOL> """<STR_LIT>""" <EOL> self . _OpenSecondaryConnection ( ) <EOL> assert _RunningInThread ( self . secondary_thread ) <EOL> assert ( not key_start ) or isinstance ( key_start , self . py_type ) , ( <EOL> '<STR_LIT>' % ( key_start , <EOL> key_start . __class__ , <EOL> self . __class__ . __name__ , <EOL> self . py_type ) ) <EOL> assert ( not key_end ) or isinstance ( key_end , self . py_type ) , ( <EOL> '<STR_LIT>' % ( key_end , <EOL> key_end . __class__ , <EOL> self . __class__ . __name__ , <EOL> self . py_type ) ) <EOL> assert KeyLEQ ( key_start , key_end ) , '<STR_LIT>' % ( <EOL> repr ( key_start ) , repr ( key_end ) ) <EOL> self . insert_cursor . execute ( <EOL> '<STR_LIT>' , <EOL> ( STATE_READ , unicode ( key_start ) , unicode ( key_end ) ) ) <EOL> progress_key = self . insert_cursor . lastrowid <EOL> self . _MaybeCommit ( ) <EOL> return progress_key <EOL> def UpdateState ( self , key , new_state ) : <EOL> """<STR_LIT>""" <EOL> self . _OpenSecondaryConnection ( ) <EOL> assert _RunningInThread ( self . secondary_thread ) <EOL> assert isinstance ( new_state , int ) <EOL> self . update_cursor . execute ( '<STR_LIT>' , <EOL> ( new_state , key ) ) <EOL> self . _MaybeCommit ( ) <EOL> def DeleteKey ( self , progress_key ) : <EOL> """<STR_LIT>""" <EOL> self . _OpenSecondaryConnection ( ) <EOL> assert _RunningInThread ( self . secondary_thread ) <EOL> t = time . time ( ) <EOL> self . insert_cursor . execute ( <EOL> '<STR_LIT>' , ( progress_key , ) ) <EOL> logger . debug ( '<STR_LIT>' , time . time ( ) - t ) <EOL> self . _MaybeCommit ( ) <EOL> def GetProgressStatusGenerator ( self ) : <EOL> """<STR_LIT>""" <EOL> conn = sqlite3 . connect ( self . db_filename , isolation_level = None ) <EOL> cursor = conn . cursor ( ) <EOL> cursor . execute ( '<STR_LIT>' ) <EOL> result = cursor . fetchone ( ) <EOL> if result is not None : <EOL> key_end = result [ <NUM_LIT:0> ] <EOL> else : <EOL> logger . debug ( '<STR_LIT>' ) <EOL> return <EOL> self . prior_key_end = key_end <EOL> cursor . execute ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> ( STATE_SENT , ) ) <EOL> rows = cursor . fetchall ( ) <EOL> for row in rows : <EOL> if row is None : <EOL> break <EOL> progress_key , state , key_start , key_end = row <EOL> yield progress_key , state , key_start , key_end <EOL> yield None , DATA_CONSUMED_TO_HERE , None , key_end <EOL> def ProgressDatabase ( db_filename , signature ) : <EOL> """<STR_LIT>""" <EOL> return _ProgressDatabase ( db_filename , '<STR_LIT>' , int , signature ) <EOL> class ExportProgressDatabase ( _ProgressDatabase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , db_filename , signature ) : <EOL> """<STR_LIT>""" <EOL> _ProgressDatabase . __init__ ( self , <EOL> db_filename , <EOL> '<STR_LIT>' , <EOL> datastore . Key , <EOL> signature , <EOL> commit_periodicity = <NUM_LIT:1> ) <EOL> def UseProgressData ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . existing_table <EOL> class StubProgressDatabase ( object ) : <EOL> """<STR_LIT>""" <EOL> def UseProgressData ( self ) : <EOL> """<STR_LIT>""" <EOL> return False <EOL> def StoreKeys ( self , unused_key_start , unused_key_end ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' <EOL> def UpdateState ( self , unused_key , unused_new_state ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def ThreadComplete ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class _ProgressThreadBase ( _ThreadBase ) : <EOL> """<STR_LIT>""" <EOL> NAME = '<STR_LIT>' <EOL> def __init__ ( self , progress_queue , progress_db ) : <EOL> """<STR_LIT>""" <EOL> _ThreadBase . __init__ ( self ) <EOL> self . progress_queue = progress_queue <EOL> self . db = progress_db <EOL> self . entities_transferred = <NUM_LIT:0> <EOL> def EntitiesTransferred ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . entities_transferred <EOL> def UpdateProgress ( self , item ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def WorkFinished ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def PerformWork ( self ) : <EOL> """<STR_LIT>""" <EOL> while not self . exit_flag : <EOL> try : <EOL> item = self . progress_queue . get ( block = True , timeout = <NUM_LIT:1.0> ) <EOL> except Queue . Empty : <EOL> continue <EOL> if item == _THREAD_SHOULD_EXIT : <EOL> break <EOL> if item . state == STATE_READ and item . progress_key is None : <EOL> item . progress_key = self . db . StoreKeys ( item . key_start , item . key_end ) <EOL> else : <EOL> assert item . progress_key is not None <EOL> self . UpdateProgress ( item ) <EOL> item . progress_event . set ( ) <EOL> self . progress_queue . task_done ( ) <EOL> self . db . ThreadComplete ( ) <EOL> class ProgressTrackerThread ( _ProgressThreadBase ) : <EOL> """<STR_LIT>""" <EOL> NAME = '<STR_LIT>' <EOL> def __init__ ( self , progress_queue , progress_db ) : <EOL> """<STR_LIT>""" <EOL> _ProgressThreadBase . __init__ ( self , progress_queue , progress_db ) <EOL> def UpdateProgress ( self , item ) : <EOL> """<STR_LIT>""" <EOL> self . db . UpdateState ( item . progress_key , item . state ) <EOL> if item . state == STATE_SENT : <EOL> self . entities_transferred += item . count <EOL> def WorkFinished ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class ExportProgressThread ( _ProgressThreadBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , kind , progress_queue , progress_db , result_db ) : <EOL> """<STR_LIT>""" <EOL> _ProgressThreadBase . __init__ ( self , progress_queue , progress_db ) <EOL> self . kind = kind <EOL> self . existing_count = result_db . existing_count <EOL> self . result_db = result_db <EOL> def EntitiesTransferred ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . result_db . count <EOL> def WorkFinished ( self ) : <EOL> """<STR_LIT>""" <EOL> exporter = Exporter . RegisteredExporter ( self . kind ) <EOL> exporter . output_entities ( self . result_db . AllEntities ( ) ) <EOL> def UpdateProgress ( self , item ) : <EOL> """<STR_LIT>""" <EOL> if item . state == STATE_GOT : <EOL> count = self . result_db . StoreEntities ( item . download_result . keys , <EOL> item . download_result . entities ) <EOL> self . db . DeleteKey ( item . progress_key ) <EOL> self . entities_transferred += count <EOL> else : <EOL> self . db . UpdateState ( item . progress_key , item . state ) <EOL> class MapperProgressThread ( _ProgressThreadBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , kind , progress_queue , progress_db ) : <EOL> """<STR_LIT>""" <EOL> _ProgressThreadBase . __init__ ( self , progress_queue , progress_db ) <EOL> self . kind = kind <EOL> self . mapper = Mapper . RegisteredMapper ( self . kind ) <EOL> def EntitiesTransferred ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . entities_transferred <EOL> def WorkFinished ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def UpdateProgress ( self , item ) : <EOL> """<STR_LIT>""" <EOL> if item . state == STATE_GOT : <EOL> self . entities_transferred += item . count <EOL> self . db . DeleteKey ( item . progress_key ) <EOL> else : <EOL> self . db . UpdateState ( item . progress_key , item . state ) <EOL> def ParseKey ( key_string ) : <EOL> """<STR_LIT>""" <EOL> if not key_string : <EOL> return None <EOL> if key_string == '<STR_LIT:None>' : <EOL> return None <EOL> return datastore . Key ( encoded = key_string ) <EOL> def Validate ( value , typ ) : <EOL> """<STR_LIT>""" <EOL> if not value : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> elif not isinstance ( value , typ ) : <EOL> raise TypeError ( '<STR_LIT>' % <EOL> ( typ , value , value . __class__ ) ) <EOL> def CheckFile ( filename ) : <EOL> """<STR_LIT>""" <EOL> if not os . path . exists ( filename ) : <EOL> raise FileNotFoundError ( '<STR_LIT>' % filename ) <EOL> elif not os . access ( filename , os . R_OK ) : <EOL> raise FileNotReadableError ( '<STR_LIT>' % filename ) <EOL> class Loader ( object ) : <EOL> """<STR_LIT>""" <EOL> __loaders = { } <EOL> kind = None <EOL> __properties = None <EOL> def __init__ ( self , kind , properties ) : <EOL> """<STR_LIT>""" <EOL> Validate ( kind , ( basestring , tuple ) ) <EOL> self . kind = kind <EOL> self . __openfile = open <EOL> self . __create_csv_reader = csv . reader <EOL> GetImplementationClass ( kind ) <EOL> Validate ( properties , list ) <EOL> for name , fn in properties : <EOL> Validate ( name , basestring ) <EOL> assert callable ( fn ) , ( <EOL> '<STR_LIT>' % ( fn , name ) ) <EOL> self . __properties = properties <EOL> @ staticmethod <EOL> def RegisterLoader ( loader ) : <EOL> """<STR_LIT>""" <EOL> Loader . __loaders [ loader . kind ] = loader <EOL> def alias_old_names ( self ) : <EOL> """<STR_LIT>""" <EOL> aliases = ( <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ) <EOL> for old_name , new_name in aliases : <EOL> setattr ( Loader , old_name , getattr ( Loader , new_name ) ) <EOL> if hasattr ( self . __class__ , old_name ) and not ( <EOL> getattr ( self . __class__ , old_name ) . im_func == <EOL> getattr ( Loader , new_name ) . im_func ) : <EOL> if hasattr ( self . __class__ , new_name ) and not ( <EOL> getattr ( self . __class__ , new_name ) . im_func == <EOL> getattr ( Loader , new_name ) . im_func ) : <EOL> raise NameClashError ( old_name , new_name , self . __class__ ) <EOL> setattr ( self , new_name , getattr ( self , old_name ) ) <EOL> def create_entity ( self , values , key_name = None , parent = None ) : <EOL> """<STR_LIT>""" <EOL> Validate ( values , ( list , tuple ) ) <EOL> assert len ( values ) == len ( self . __properties ) , ( <EOL> '<STR_LIT>' % <EOL> ( len ( self . __properties ) , len ( values ) ) ) <EOL> model_class = GetImplementationClass ( self . kind ) <EOL> properties = { <EOL> '<STR_LIT>' : key_name , <EOL> '<STR_LIT>' : parent , <EOL> } <EOL> for ( name , converter ) , val in zip ( self . __properties , values ) : <EOL> if converter is bool and val . lower ( ) in ( '<STR_LIT:0>' , '<STR_LIT:false>' , '<STR_LIT>' ) : <EOL> val = False <EOL> properties [ name ] = converter ( val ) <EOL> entity = model_class ( ** properties ) <EOL> entities = self . handle_entity ( entity ) <EOL> if entities : <EOL> if not isinstance ( entities , ( list , tuple ) ) : <EOL> entities = [ entities ] <EOL> for entity in entities : <EOL> if not isinstance ( entity , db . Model ) : <EOL> raise TypeError ( '<STR_LIT>' % <EOL> ( entity , entity . __class__ ) ) <EOL> return entities <EOL> def generate_key ( self , i , values ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def handle_entity ( self , entity ) : <EOL> """<STR_LIT>""" <EOL> return entity <EOL> def initialize ( self , filename , loader_opts ) : <EOL> """<STR_LIT>""" <EOL> CheckFile ( filename ) <EOL> def finalize ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def generate_records ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> csv_generator = CSVGenerator ( filename , openfile = self . __openfile , <EOL> create_csv_reader = self . __create_csv_reader <EOL> ) . Records ( ) <EOL> return csv_generator <EOL> @ staticmethod <EOL> def RegisteredLoaders ( ) : <EOL> """<STR_LIT>""" <EOL> return dict ( Loader . __loaders ) <EOL> @ staticmethod <EOL> def RegisteredLoader ( kind ) : <EOL> """<STR_LIT>""" <EOL> return Loader . __loaders [ kind ] <EOL> class RestoreThread ( _ThreadBase ) : <EOL> """<STR_LIT>""" <EOL> NAME = '<STR_LIT>' <EOL> _ENTITIES_DONE = '<STR_LIT>' <EOL> def __init__ ( self , queue , filename ) : <EOL> _ThreadBase . __init__ ( self ) <EOL> self . queue = queue <EOL> self . filename = filename <EOL> def PerformWork ( self ) : <EOL> db_conn = sqlite3 . connect ( self . filename ) <EOL> cursor = db_conn . cursor ( ) <EOL> cursor . execute ( '<STR_LIT>' ) <EOL> for entity_id , value in cursor : <EOL> self . queue . put ( [ entity_id , value ] , block = True ) <EOL> self . queue . put ( RestoreThread . _ENTITIES_DONE , block = True ) <EOL> class RestoreLoader ( Loader ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , kind , app_id ) : <EOL> self . kind = kind <EOL> self . app_id = app_id <EOL> def initialize ( self , filename , loader_opts ) : <EOL> CheckFile ( filename ) <EOL> self . queue = Queue . Queue ( <NUM_LIT:1000> ) <EOL> restore_thread = RestoreThread ( self . queue , filename ) <EOL> restore_thread . start ( ) <EOL> def generate_records ( self , filename ) : <EOL> while True : <EOL> record = self . queue . get ( block = True ) <EOL> if id ( record ) == id ( RestoreThread . _ENTITIES_DONE ) : <EOL> break <EOL> yield record <EOL> def create_entity ( self , values , key_name = None , parent = None ) : <EOL> def convert_key ( key , app_id ) : <EOL> path = key . to_path ( ) <EOL> kwargs = { '<STR_LIT>' : app_id } <EOL> return db . Key . from_path ( * path , ** kwargs ) <EOL> import copy <EOL> key = StrKey ( unicode ( values [ <NUM_LIT:0> ] , '<STR_LIT:utf-8>' ) ) <EOL> entity_proto = entity_pb . EntityProto ( contents = str ( values [ <NUM_LIT:1> ] ) ) <EOL> entity_proto . mutable_key ( ) . CopyFrom ( key . _Key__reference ) <EOL> entity = datastore . Entity . _FromPb ( entity_proto ) <EOL> new_entity = copy . copy ( entity ) <EOL> for k , v in entity . iteritems ( ) : <EOL> if isinstance ( v , db . Key ) : <EOL> new_entity [ k ] = convert_key ( v , self . app_id ) <EOL> if isinstance ( v , list ) : <EOL> new_list = [ ] <EOL> for item in v : <EOL> if isinstance ( item , db . Key ) : <EOL> new_list . append ( convert_key ( item , self . app_id ) ) <EOL> else : <EOL> new_list . append ( item ) <EOL> new_entity [ k ] = new_list <EOL> return new_entity <EOL> class Exporter ( object ) : <EOL> """<STR_LIT>""" <EOL> __exporters = { } <EOL> kind = None <EOL> __properties = None <EOL> def __init__ ( self , kind , properties ) : <EOL> """<STR_LIT>""" <EOL> Validate ( kind , basestring ) <EOL> self . kind = kind <EOL> GetImplementationClass ( kind ) <EOL> Validate ( properties , list ) <EOL> for name , fn , default in properties : <EOL> Validate ( name , basestring ) <EOL> assert callable ( fn ) , ( <EOL> '<STR_LIT>' % ( <EOL> fn , name ) ) <EOL> if default : <EOL> Validate ( default , basestring ) <EOL> self . __properties = properties <EOL> @ staticmethod <EOL> def RegisterExporter ( exporter ) : <EOL> """<STR_LIT>""" <EOL> Exporter . __exporters [ exporter . kind ] = exporter <EOL> def __ExtractProperties ( self , entity ) : <EOL> """<STR_LIT>""" <EOL> encoding = [ ] <EOL> for name , fn , default in self . __properties : <EOL> try : <EOL> encoding . append ( fn ( entity [ name ] ) ) <EOL> except AttributeError : <EOL> if default is None : <EOL> raise MissingPropertyError ( name ) <EOL> else : <EOL> encoding . append ( default ) <EOL> return encoding <EOL> def __EncodeEntity ( self , entity ) : <EOL> """<STR_LIT>""" <EOL> output = StringIO . StringIO ( ) <EOL> writer = csv . writer ( output , lineterminator = '<STR_LIT>' ) <EOL> writer . writerow ( self . __ExtractProperties ( entity ) ) <EOL> return output . getvalue ( ) <EOL> def __SerializeEntity ( self , entity ) : <EOL> """<STR_LIT>""" <EOL> encoding = self . __EncodeEntity ( entity ) <EOL> if not isinstance ( encoding , unicode ) : <EOL> encoding = unicode ( encoding , '<STR_LIT:utf-8>' ) <EOL> encoding = encoding . encode ( '<STR_LIT:utf-8>' ) <EOL> return encoding <EOL> def output_entities ( self , entity_generator ) : <EOL> """<STR_LIT>""" <EOL> CheckOutputFile ( self . output_filename ) <EOL> output_file = open ( self . output_filename , '<STR_LIT:w>' ) <EOL> logger . debug ( '<STR_LIT>' ) <EOL> output_file . writelines ( self . __SerializeEntity ( entity ) + '<STR_LIT:\n>' <EOL> for entity in entity_generator ) <EOL> def initialize ( self , filename , exporter_opts ) : <EOL> """<STR_LIT>""" <EOL> CheckOutputFile ( filename ) <EOL> self . output_filename = filename <EOL> def finalize ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ staticmethod <EOL> def RegisteredExporters ( ) : <EOL> """<STR_LIT>""" <EOL> return dict ( Exporter . __exporters ) <EOL> @ staticmethod <EOL> def RegisteredExporter ( kind ) : <EOL> """<STR_LIT>""" <EOL> return Exporter . __exporters [ kind ] <EOL> class DumpExporter ( Exporter ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , kind , result_db_filename ) : <EOL> self . kind = kind <EOL> self . result_db_filename = result_db_filename <EOL> def output_entities ( self , entity_generator ) : <EOL> shutil . copyfile ( self . result_db_filename , self . output_filename ) <EOL> class MapperRetry ( Error ) : <EOL> """<STR_LIT>""" <EOL> class Mapper ( object ) : <EOL> """<STR_LIT>""" <EOL> __mappers = { } <EOL> kind = None <EOL> def __init__ ( self , kind ) : <EOL> """<STR_LIT>""" <EOL> Validate ( kind , basestring ) <EOL> self . kind = kind <EOL> GetImplementationClass ( kind ) <EOL> @ staticmethod <EOL> def RegisterMapper ( mapper ) : <EOL> """<STR_LIT>""" <EOL> Mapper . __mappers [ mapper . kind ] = mapper <EOL> def initialize ( self , mapper_opts ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def finalize ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def apply ( self , entity ) : <EOL> print '<STR_LIT>' % entity <EOL> def batch_apply ( self , entities ) : <EOL> for entity in entities : <EOL> self . apply ( entity ) <EOL> @ staticmethod <EOL> def RegisteredMappers ( ) : <EOL> """<STR_LIT>""" <EOL> return dict ( Mapper . __mappers ) <EOL> @ staticmethod <EOL> def RegisteredMapper ( kind ) : <EOL> """<STR_LIT>""" <EOL> return Mapper . __mappers [ kind ] <EOL> class QueueJoinThread ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , queue ) : <EOL> """<STR_LIT>""" <EOL> threading . Thread . __init__ ( self ) <EOL> assert isinstance ( queue , ( Queue . Queue , ReQueue ) ) <EOL> self . queue = queue <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> self . queue . join ( ) <EOL> def InterruptibleQueueJoin ( queue , <EOL> thread_local , <EOL> thread_pool , <EOL> queue_join_thread_factory = QueueJoinThread , <EOL> check_workers = True ) : <EOL> """<STR_LIT>""" <EOL> thread = queue_join_thread_factory ( queue ) <EOL> thread . start ( ) <EOL> while True : <EOL> thread . join ( timeout = <NUM_LIT> ) <EOL> if not thread . isAlive ( ) : <EOL> return True <EOL> if thread_local . shut_down : <EOL> logger . debug ( '<STR_LIT>' ) <EOL> return False <EOL> if check_workers : <EOL> for worker_thread in thread_pool . Threads ( ) : <EOL> if not worker_thread . isAlive ( ) : <EOL> return False <EOL> def ShutdownThreads ( data_source_thread , thread_pool ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( '<STR_LIT>' ) <EOL> data_source_thread . exit_flag = True <EOL> thread_pool . Shutdown ( ) <EOL> data_source_thread . join ( timeout = <NUM_LIT> ) <EOL> if data_source_thread . isAlive ( ) : <EOL> logger . warn ( '<STR_LIT>' , <EOL> data_source_thread . GetFriendlyName ( ) ) <EOL> class BulkTransporterApp ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> arg_dict , <EOL> input_generator_factory , <EOL> throttle , <EOL> progress_db , <EOL> progresstrackerthread_factory , <EOL> max_queue_size = DEFAULT_QUEUE_SIZE , <EOL> request_manager_factory = RequestManager , <EOL> datasourcethread_factory = DataSourceThread , <EOL> progress_queue_factory = Queue . Queue , <EOL> thread_pool_factory = adaptive_thread_pool . AdaptiveThreadPool ) : <EOL> """<STR_LIT>""" <EOL> self . app_id = arg_dict [ '<STR_LIT>' ] <EOL> self . post_url = arg_dict [ '<STR_LIT:url>' ] <EOL> self . kind = arg_dict [ '<STR_LIT>' ] <EOL> self . batch_size = arg_dict [ '<STR_LIT>' ] <EOL> self . input_generator_factory = input_generator_factory <EOL> self . num_threads = arg_dict [ '<STR_LIT>' ] <EOL> self . email = arg_dict [ '<STR_LIT:email>' ] <EOL> self . passin = arg_dict [ '<STR_LIT>' ] <EOL> self . dry_run = arg_dict [ '<STR_LIT>' ] <EOL> self . throttle = throttle <EOL> self . progress_db = progress_db <EOL> self . progresstrackerthread_factory = progresstrackerthread_factory <EOL> self . max_queue_size = max_queue_size <EOL> self . request_manager_factory = request_manager_factory <EOL> self . datasourcethread_factory = datasourcethread_factory <EOL> self . progress_queue_factory = progress_queue_factory <EOL> self . thread_pool_factory = thread_pool_factory <EOL> ( scheme , <EOL> self . host_port , self . url_path , <EOL> unused_query , unused_fragment ) = urlparse . urlsplit ( self . post_url ) <EOL> self . secure = ( scheme == '<STR_LIT>' ) <EOL> def Run ( self ) : <EOL> """<STR_LIT>""" <EOL> self . error = False <EOL> thread_pool = self . thread_pool_factory ( <EOL> self . num_threads , queue_size = self . max_queue_size ) <EOL> self . throttle . Register ( threading . currentThread ( ) ) <EOL> threading . currentThread ( ) . exit_flag = False <EOL> progress_queue = self . progress_queue_factory ( self . max_queue_size ) <EOL> request_manager = self . request_manager_factory ( self . app_id , <EOL> self . host_port , <EOL> self . url_path , <EOL> self . kind , <EOL> self . throttle , <EOL> self . batch_size , <EOL> self . secure , <EOL> self . email , <EOL> self . passin , <EOL> self . dry_run ) <EOL> try : <EOL> request_manager . Authenticate ( ) <EOL> except Exception , e : <EOL> self . error = True <EOL> if not isinstance ( e , urllib2 . HTTPError ) or ( <EOL> e . code != <NUM_LIT> and e . code != <NUM_LIT> ) : <EOL> logger . exception ( '<STR_LIT>' ) <EOL> raise AuthenticationError ( ) <EOL> if ( request_manager . auth_called and <EOL> not request_manager . authenticated ) : <EOL> self . error = True <EOL> raise AuthenticationError ( '<STR_LIT>' ) <EOL> for thread in thread_pool . Threads ( ) : <EOL> self . throttle . Register ( thread ) <EOL> self . progress_thread = self . progresstrackerthread_factory ( <EOL> progress_queue , self . progress_db ) <EOL> if self . progress_db . UseProgressData ( ) : <EOL> logger . debug ( '<STR_LIT>' ) <EOL> progress_generator_factory = self . progress_db . GetProgressStatusGenerator <EOL> else : <EOL> progress_generator_factory = None <EOL> self . data_source_thread = ( <EOL> self . datasourcethread_factory ( request_manager , <EOL> thread_pool , <EOL> progress_queue , <EOL> self . input_generator_factory , <EOL> progress_generator_factory ) ) <EOL> thread_local = threading . local ( ) <EOL> thread_local . shut_down = False <EOL> def Interrupt ( unused_signum , unused_frame ) : <EOL> """<STR_LIT>""" <EOL> thread_local . shut_down = True <EOL> self . error = True <EOL> signal . signal ( signal . SIGINT , Interrupt ) <EOL> self . progress_thread . start ( ) <EOL> self . data_source_thread . start ( ) <EOL> while not thread_local . shut_down : <EOL> self . data_source_thread . join ( timeout = <NUM_LIT> ) <EOL> if self . data_source_thread . isAlive ( ) : <EOL> for thread in list ( thread_pool . Threads ( ) ) + [ self . progress_thread ] : <EOL> if not thread . isAlive ( ) : <EOL> logger . info ( '<STR_LIT>' , thread . getName ( ) ) <EOL> thread_local . shut_down = True <EOL> self . error = True <EOL> break <EOL> else : <EOL> break <EOL> def _Join ( ob , msg ) : <EOL> logger . debug ( '<STR_LIT>' , msg ) <EOL> if isinstance ( ob , threading . Thread ) : <EOL> ob . join ( timeout = <NUM_LIT> ) <EOL> if ob . isAlive ( ) : <EOL> logger . debug ( '<STR_LIT>' , ob ) <EOL> else : <EOL> logger . debug ( '<STR_LIT>' ) <EOL> elif isinstance ( ob , ( Queue . Queue , ReQueue ) ) : <EOL> if not InterruptibleQueueJoin ( ob , thread_local , thread_pool ) : <EOL> ShutdownThreads ( self . data_source_thread , thread_pool ) <EOL> else : <EOL> ob . join ( ) <EOL> logger . debug ( '<STR_LIT>' ) <EOL> if self . data_source_thread . error or thread_local . shut_down : <EOL> ShutdownThreads ( self . data_source_thread , thread_pool ) <EOL> else : <EOL> _Join ( thread_pool . requeue , '<STR_LIT>' ) <EOL> thread_pool . Shutdown ( ) <EOL> thread_pool . JoinThreads ( ) <EOL> thread_pool . CheckErrors ( ) <EOL> print '<STR_LIT>' <EOL> if self . progress_thread . isAlive ( ) : <EOL> InterruptibleQueueJoin ( progress_queue , thread_local , thread_pool , <EOL> check_workers = False ) <EOL> else : <EOL> logger . warn ( '<STR_LIT>' ) <EOL> progress_queue . put ( _THREAD_SHOULD_EXIT ) <EOL> _Join ( self . progress_thread , '<STR_LIT>' ) <EOL> self . progress_thread . CheckError ( ) <EOL> if not thread_local . shut_down : <EOL> self . progress_thread . WorkFinished ( ) <EOL> self . data_source_thread . CheckError ( ) <EOL> return self . ReportStatus ( ) <EOL> def ReportStatus ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> class BulkUploaderApp ( BulkTransporterApp ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> BulkTransporterApp . __init__ ( self , * args , ** kwargs ) <EOL> def ReportStatus ( self ) : <EOL> """<STR_LIT>""" <EOL> total_up , duration = self . throttle . TotalTransferred ( <EOL> remote_api_throttle . BANDWIDTH_UP ) <EOL> s_total_up , unused_duration = self . throttle . TotalTransferred ( <EOL> remote_api_throttle . HTTPS_BANDWIDTH_UP ) <EOL> total_up += s_total_up <EOL> total = total_up <EOL> logger . info ( '<STR_LIT>' , <EOL> self . data_source_thread . read_count , <EOL> self . data_source_thread . xfer_count ) <EOL> transfer_count = self . progress_thread . EntitiesTransferred ( ) <EOL> logger . info ( '<STR_LIT>' , <EOL> transfer_count , total , duration ) <EOL> if ( self . data_source_thread . read_all and <EOL> transfer_count + <EOL> self . data_source_thread . xfer_count >= <EOL> self . data_source_thread . read_count ) : <EOL> logger . info ( '<STR_LIT>' ) <EOL> return <NUM_LIT:0> <EOL> else : <EOL> logger . info ( '<STR_LIT>' ) <EOL> return <NUM_LIT:1> <EOL> class BulkDownloaderApp ( BulkTransporterApp ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> BulkTransporterApp . __init__ ( self , * args , ** kwargs ) <EOL> def ReportStatus ( self ) : <EOL> """<STR_LIT>""" <EOL> total_down , duration = self . throttle . TotalTransferred ( <EOL> remote_api_throttle . BANDWIDTH_DOWN ) <EOL> s_total_down , unused_duration = self . throttle . TotalTransferred ( <EOL> remote_api_throttle . HTTPS_BANDWIDTH_DOWN ) <EOL> total_down += s_total_down <EOL> total = total_down <EOL> existing_count = self . progress_thread . existing_count <EOL> xfer_count = self . progress_thread . EntitiesTransferred ( ) <EOL> logger . info ( '<STR_LIT>' , <EOL> xfer_count , existing_count ) <EOL> logger . info ( '<STR_LIT>' , <EOL> xfer_count , total , duration ) <EOL> if self . error : <EOL> return <NUM_LIT:1> <EOL> else : <EOL> return <NUM_LIT:0> <EOL> class BulkMapperApp ( BulkTransporterApp ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> BulkTransporterApp . __init__ ( self , * args , ** kwargs ) <EOL> def ReportStatus ( self ) : <EOL> """<STR_LIT>""" <EOL> total_down , duration = self . throttle . TotalTransferred ( <EOL> remote_api_throttle . BANDWIDTH_DOWN ) <EOL> s_total_down , unused_duration = self . throttle . TotalTransferred ( <EOL> remote_api_throttle . HTTPS_BANDWIDTH_DOWN ) <EOL> total_down += s_total_down <EOL> total = total_down <EOL> xfer_count = self . progress_thread . EntitiesTransferred ( ) <EOL> logger . info ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> logger . info ( '<STR_LIT>' , <EOL> xfer_count ) <EOL> logger . info ( '<STR_LIT>' , <EOL> xfer_count , total , duration ) <EOL> if self . error : <EOL> return <NUM_LIT:1> <EOL> else : <EOL> return <NUM_LIT:0> <EOL> def PrintUsageExit ( code ) : <EOL> """<STR_LIT>""" <EOL> print __doc__ % { '<STR_LIT>' : sys . argv [ <NUM_LIT:0> ] } <EOL> sys . stdout . flush ( ) <EOL> sys . stderr . flush ( ) <EOL> sys . exit ( code ) <EOL> REQUIRED_OPTION = object ( ) <EOL> FLAG_SPEC = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> def ParseArguments ( argv , die_fn = lambda : PrintUsageExit ( <NUM_LIT:1> ) ) : <EOL> """<STR_LIT>""" <EOL> opts , unused_args = getopt . getopt ( <EOL> argv [ <NUM_LIT:1> : ] , <EOL> '<STR_LIT:h>' , <EOL> FLAG_SPEC ) <EOL> arg_dict = { } <EOL> arg_dict [ '<STR_LIT:url>' ] = REQUIRED_OPTION <EOL> arg_dict [ '<STR_LIT:filename>' ] = None <EOL> arg_dict [ '<STR_LIT>' ] = None <EOL> arg_dict [ '<STR_LIT>' ] = None <EOL> arg_dict [ '<STR_LIT>' ] = None <EOL> arg_dict [ '<STR_LIT>' ] = DEFAULT_THREAD_COUNT <EOL> arg_dict [ '<STR_LIT>' ] = DEFAULT_BANDWIDTH_LIMIT <EOL> arg_dict [ '<STR_LIT>' ] = DEFAULT_RPS_LIMIT <EOL> arg_dict [ '<STR_LIT>' ] = DEFAULT_REQUEST_LIMIT <EOL> arg_dict [ '<STR_LIT>' ] = None <EOL> arg_dict [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> arg_dict [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> arg_dict [ '<STR_LIT>' ] = False <EOL> arg_dict [ '<STR_LIT>' ] = None <EOL> arg_dict [ '<STR_LIT>' ] = False <EOL> arg_dict [ '<STR_LIT>' ] = None <EOL> arg_dict [ '<STR_LIT>' ] = None <EOL> arg_dict [ '<STR_LIT>' ] = False <EOL> arg_dict [ '<STR_LIT>' ] = None <EOL> arg_dict [ '<STR_LIT:email>' ] = None <EOL> arg_dict [ '<STR_LIT>' ] = False <EOL> arg_dict [ '<STR_LIT>' ] = None <EOL> arg_dict [ '<STR_LIT>' ] = False <EOL> arg_dict [ '<STR_LIT>' ] = False <EOL> arg_dict [ '<STR_LIT>' ] = False <EOL> arg_dict [ '<STR_LIT>' ] = False <EOL> def ExpandFilename ( filename ) : <EOL> """<STR_LIT>""" <EOL> return os . path . expandvars ( os . path . expanduser ( filename ) ) <EOL> for option , value in opts : <EOL> if option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = True <EOL> elif option in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> PrintUsageExit ( <NUM_LIT:0> ) <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT:url>' ] = value <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT:filename>' ] = ExpandFilename ( value ) <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = int ( value ) <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = value <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = int ( value ) <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = int ( value ) <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = int ( value ) <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = int ( value ) <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = ExpandFilename ( value ) <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = value <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = ExpandFilename ( value ) <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = value <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = True <EOL> elif option == '<STR_LIT>' : <EOL> print >> sys . stderr , ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> arg_dict [ '<STR_LIT>' ] = True <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = ExpandFilename ( value ) <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = True <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = value <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = value <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = ExpandFilename ( value ) <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT:email>' ] = value <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = True <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = True <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = value <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = True <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = True <EOL> elif option == '<STR_LIT>' : <EOL> arg_dict [ '<STR_LIT>' ] = True <EOL> return ProcessArguments ( arg_dict , die_fn = die_fn ) <EOL> def ThrottleLayout ( bandwidth_limit , http_limit , rps_limit ) : <EOL> """<STR_LIT>""" <EOL> bulkloader_limits = dict ( remote_api_throttle . NO_LIMITS ) <EOL> bulkloader_limits . update ( { <EOL> remote_api_throttle . BANDWIDTH_UP : bandwidth_limit , <EOL> remote_api_throttle . BANDWIDTH_DOWN : bandwidth_limit , <EOL> remote_api_throttle . REQUESTS : http_limit , <EOL> remote_api_throttle . HTTPS_BANDWIDTH_UP : bandwidth_limit , <EOL> remote_api_throttle . HTTPS_BANDWIDTH_DOWN : bandwidth_limit , <EOL> remote_api_throttle . HTTPS_REQUESTS : http_limit , <EOL> remote_api_throttle . ENTITIES_FETCHED : rps_limit , <EOL> remote_api_throttle . ENTITIES_MODIFIED : rps_limit , <EOL> } ) <EOL> return bulkloader_limits <EOL> def CheckOutputFile ( filename ) : <EOL> """<STR_LIT>""" <EOL> full_path = os . path . abspath ( filename ) <EOL> if os . path . exists ( full_path ) : <EOL> raise FileExistsError ( '<STR_LIT>' % filename ) <EOL> elif not os . access ( os . path . dirname ( full_path ) , os . W_OK ) : <EOL> raise FileNotWritableError ( <EOL> '<STR_LIT>' % os . path . dirname ( full_path ) ) <EOL> def LoadConfig ( config_file_name , exit_fn = sys . exit ) : <EOL> """<STR_LIT>""" <EOL> if config_file_name : <EOL> config_file = open ( config_file_name , '<STR_LIT:r>' ) <EOL> try : <EOL> bulkloader_config = imp . load_module ( <EOL> '<STR_LIT>' , config_file , config_file_name , <EOL> ( '<STR_LIT>' , '<STR_LIT:r>' , imp . PY_SOURCE ) ) <EOL> sys . modules [ '<STR_LIT>' ] = bulkloader_config <EOL> if hasattr ( bulkloader_config , '<STR_LIT>' ) : <EOL> for cls in bulkloader_config . loaders : <EOL> Loader . RegisterLoader ( cls ( ) ) <EOL> if hasattr ( bulkloader_config , '<STR_LIT>' ) : <EOL> for cls in bulkloader_config . exporters : <EOL> Exporter . RegisterExporter ( cls ( ) ) <EOL> if hasattr ( bulkloader_config , '<STR_LIT>' ) : <EOL> for cls in bulkloader_config . mappers : <EOL> Mapper . RegisterMapper ( cls ( ) ) <EOL> except NameError , e : <EOL> m = re . search ( r"<STR_LIT>" , str ( e ) ) <EOL> if m . groups ( ) and m . group ( <NUM_LIT:1> ) == '<STR_LIT>' : <EOL> print >> sys . stderr , """<STR_LIT>""" <EOL> exit_fn ( <NUM_LIT:1> ) <EOL> else : <EOL> raise <EOL> except Exception , e : <EOL> if isinstance ( e , NameClashError ) or '<STR_LIT>' in vars ( ) and ( <EOL> hasattr ( bulkloader_config , '<STR_LIT>' ) and <EOL> isinstance ( e , bulkloader_config . bulkloader . NameClashError ) ) : <EOL> print >> sys . stderr , ( <EOL> '<STR_LIT>' % <EOL> ( e . old_name , e . new_name , e . klass ) ) <EOL> exit_fn ( <NUM_LIT:1> ) <EOL> else : <EOL> raise <EOL> def GetArgument ( kwargs , name , die_fn ) : <EOL> """<STR_LIT>""" <EOL> if name in kwargs : <EOL> return kwargs [ name ] <EOL> else : <EOL> print >> sys . stderr , '<STR_LIT>' % name <EOL> die_fn ( ) <EOL> def _MakeSignature ( app_id = None , <EOL> url = None , <EOL> kind = None , <EOL> db_filename = None , <EOL> perform_map = None , <EOL> download = None , <EOL> has_header = None , <EOL> result_db_filename = None , <EOL> dump = None , <EOL> restore = None ) : <EOL> """<STR_LIT>""" <EOL> if download : <EOL> result_db_line = '<STR_LIT>' % result_db_filename <EOL> else : <EOL> result_db_line = '<STR_LIT>' <EOL> return u"""<STR_LIT>""" % ( app_id , url , kind , download , perform_map , dump , restore , db_filename , <EOL> has_header , result_db_line ) <EOL> def ProcessArguments ( arg_dict , <EOL> die_fn = lambda : sys . exit ( <NUM_LIT:1> ) ) : <EOL> """<STR_LIT>""" <EOL> app_id = GetArgument ( arg_dict , '<STR_LIT>' , die_fn ) <EOL> url = GetArgument ( arg_dict , '<STR_LIT:url>' , die_fn ) <EOL> dump = GetArgument ( arg_dict , '<STR_LIT>' , die_fn ) <EOL> restore = GetArgument ( arg_dict , '<STR_LIT>' , die_fn ) <EOL> filename = GetArgument ( arg_dict , '<STR_LIT:filename>' , die_fn ) <EOL> batch_size = GetArgument ( arg_dict , '<STR_LIT>' , die_fn ) <EOL> kind = GetArgument ( arg_dict , '<STR_LIT>' , die_fn ) <EOL> db_filename = GetArgument ( arg_dict , '<STR_LIT>' , die_fn ) <EOL> config_file = GetArgument ( arg_dict , '<STR_LIT>' , die_fn ) <EOL> result_db_filename = GetArgument ( arg_dict , '<STR_LIT>' , die_fn ) <EOL> download = GetArgument ( arg_dict , '<STR_LIT>' , die_fn ) <EOL> log_file = GetArgument ( arg_dict , '<STR_LIT>' , die_fn ) <EOL> perform_map = GetArgument ( arg_dict , '<STR_LIT>' , die_fn ) <EOL> errors = [ ] <EOL> if batch_size is None : <EOL> if download or perform_map : <EOL> arg_dict [ '<STR_LIT>' ] = DEFAULT_DOWNLOAD_BATCH_SIZE <EOL> else : <EOL> arg_dict [ '<STR_LIT>' ] = DEFAULT_BATCH_SIZE <EOL> elif batch_size <= <NUM_LIT:0> : <EOL> errors . append ( '<STR_LIT>' ) <EOL> if db_filename is None : <EOL> arg_dict [ '<STR_LIT>' ] = time . strftime ( <EOL> '<STR_LIT>' ) <EOL> if result_db_filename is None : <EOL> arg_dict [ '<STR_LIT>' ] = time . strftime ( <EOL> '<STR_LIT>' ) <EOL> if log_file is None : <EOL> arg_dict [ '<STR_LIT>' ] = time . strftime ( '<STR_LIT>' ) <EOL> required = '<STR_LIT>' <EOL> if config_file is None and not dump and not restore : <EOL> errors . append ( '<STR_LIT>' ) <EOL> if url is REQUIRED_OPTION : <EOL> errors . append ( required % '<STR_LIT:url>' ) <EOL> if not filename and not perform_map : <EOL> errors . append ( required % '<STR_LIT:filename>' ) <EOL> if kind is None : <EOL> if download or map : <EOL> errors . append ( '<STR_LIT>' ) <EOL> elif not dump and not restore : <EOL> errors . append ( <EOL> '<STR_LIT>' ) <EOL> if not app_id : <EOL> if url and url is not REQUIRED_OPTION : <EOL> ( unused_scheme , host_port , unused_url_path , <EOL> unused_query , unused_fragment ) = urlparse . urlsplit ( url ) <EOL> suffix_idx = host_port . find ( '<STR_LIT>' ) <EOL> if suffix_idx > - <NUM_LIT:1> : <EOL> arg_dict [ '<STR_LIT>' ] = host_port [ : suffix_idx ] <EOL> elif host_port . split ( '<STR_LIT::>' ) [ <NUM_LIT:0> ] . endswith ( '<STR_LIT>' ) : <EOL> arg_dict [ '<STR_LIT>' ] = host_port . split ( '<STR_LIT:.>' ) [ <NUM_LIT:0> ] <EOL> else : <EOL> errors . append ( '<STR_LIT>' ) <EOL> if errors : <EOL> print >> sys . stderr , '<STR_LIT:\n>' . join ( errors ) <EOL> die_fn ( ) <EOL> return arg_dict <EOL> def ParseKind ( kind ) : <EOL> if kind and kind [ <NUM_LIT:0> ] == '<STR_LIT:(>' and kind [ - <NUM_LIT:1> ] == '<STR_LIT:)>' : <EOL> return tuple ( kind [ <NUM_LIT:1> : - <NUM_LIT:1> ] . split ( '<STR_LIT:U+002C>' ) ) <EOL> else : <EOL> return kind <EOL> def _PerformBulkload ( arg_dict , <EOL> check_file = CheckFile , <EOL> check_output_file = CheckOutputFile ) : <EOL> """<STR_LIT>""" <EOL> app_id = arg_dict [ '<STR_LIT>' ] <EOL> url = arg_dict [ '<STR_LIT:url>' ] <EOL> filename = arg_dict [ '<STR_LIT:filename>' ] <EOL> batch_size = arg_dict [ '<STR_LIT>' ] <EOL> kind = arg_dict [ '<STR_LIT>' ] <EOL> num_threads = arg_dict [ '<STR_LIT>' ] <EOL> bandwidth_limit = arg_dict [ '<STR_LIT>' ] <EOL> rps_limit = arg_dict [ '<STR_LIT>' ] <EOL> http_limit = arg_dict [ '<STR_LIT>' ] <EOL> db_filename = arg_dict [ '<STR_LIT>' ] <EOL> config_file = arg_dict [ '<STR_LIT>' ] <EOL> auth_domain = arg_dict [ '<STR_LIT>' ] <EOL> has_header = arg_dict [ '<STR_LIT>' ] <EOL> download = arg_dict [ '<STR_LIT>' ] <EOL> result_db_filename = arg_dict [ '<STR_LIT>' ] <EOL> loader_opts = arg_dict [ '<STR_LIT>' ] <EOL> exporter_opts = arg_dict [ '<STR_LIT>' ] <EOL> mapper_opts = arg_dict [ '<STR_LIT>' ] <EOL> email = arg_dict [ '<STR_LIT:email>' ] <EOL> passin = arg_dict [ '<STR_LIT>' ] <EOL> perform_map = arg_dict [ '<STR_LIT>' ] <EOL> dump = arg_dict [ '<STR_LIT>' ] <EOL> restore = arg_dict [ '<STR_LIT>' ] <EOL> os . environ [ '<STR_LIT>' ] = auth_domain <EOL> kind = ParseKind ( kind ) <EOL> if not dump and not restore : <EOL> check_file ( config_file ) <EOL> if download and perform_map : <EOL> logger . error ( '<STR_LIT>' ) <EOL> if download or dump : <EOL> check_output_file ( filename ) <EOL> elif not perform_map : <EOL> check_file ( filename ) <EOL> if dump : <EOL> Exporter . RegisterExporter ( DumpExporter ( kind , result_db_filename ) ) <EOL> elif restore : <EOL> Loader . RegisterLoader ( RestoreLoader ( kind , app_id ) ) <EOL> else : <EOL> LoadConfig ( config_file ) <EOL> os . environ [ '<STR_LIT>' ] = app_id <EOL> throttle_layout = ThrottleLayout ( bandwidth_limit , http_limit , rps_limit ) <EOL> logger . info ( '<STR_LIT>' ) <EOL> logger . info ( '<STR_LIT>' , bandwidth_limit ) <EOL> logger . info ( '<STR_LIT>' , http_limit ) <EOL> logger . info ( '<STR_LIT>' , rps_limit ) <EOL> throttle = remote_api_throttle . Throttle ( layout = throttle_layout ) <EOL> signature = _MakeSignature ( app_id = app_id , <EOL> url = url , <EOL> kind = kind , <EOL> db_filename = db_filename , <EOL> download = download , <EOL> perform_map = perform_map , <EOL> has_header = has_header , <EOL> result_db_filename = result_db_filename , <EOL> dump = dump , <EOL> restore = restore ) <EOL> max_queue_size = max ( DEFAULT_QUEUE_SIZE , <NUM_LIT:3> * num_threads + <NUM_LIT:5> ) <EOL> if db_filename == '<STR_LIT>' : <EOL> progress_db = StubProgressDatabase ( ) <EOL> elif not download and not perform_map and not dump : <EOL> progress_db = ProgressDatabase ( db_filename , signature ) <EOL> else : <EOL> progress_db = ExportProgressDatabase ( db_filename , signature ) <EOL> return_code = <NUM_LIT:1> <EOL> if not download and not perform_map and not dump : <EOL> loader = Loader . RegisteredLoader ( kind ) <EOL> try : <EOL> loader . initialize ( filename , loader_opts ) <EOL> workitem_generator_factory = GetCSVGeneratorFactory ( <EOL> kind , filename , batch_size , has_header ) <EOL> app = BulkUploaderApp ( arg_dict , <EOL> workitem_generator_factory , <EOL> throttle , <EOL> progress_db , <EOL> ProgressTrackerThread , <EOL> max_queue_size , <EOL> RequestManager , <EOL> DataSourceThread , <EOL> Queue . Queue ) <EOL> try : <EOL> return_code = app . Run ( ) <EOL> except AuthenticationError : <EOL> logger . info ( '<STR_LIT>' ) <EOL> finally : <EOL> loader . finalize ( ) <EOL> elif not perform_map : <EOL> result_db = ResultDatabase ( result_db_filename , signature ) <EOL> exporter = Exporter . RegisteredExporter ( kind ) <EOL> try : <EOL> exporter . initialize ( filename , exporter_opts ) <EOL> def KeyRangeGeneratorFactory ( request_manager , progress_queue , <EOL> progress_gen ) : <EOL> return KeyRangeItemGenerator ( request_manager , kind , progress_queue , <EOL> progress_gen , DownloadItem ) <EOL> def ExportProgressThreadFactory ( progress_queue , progress_db ) : <EOL> return ExportProgressThread ( kind , <EOL> progress_queue , <EOL> progress_db , <EOL> result_db ) <EOL> app = BulkDownloaderApp ( arg_dict , <EOL> KeyRangeGeneratorFactory , <EOL> throttle , <EOL> progress_db , <EOL> ExportProgressThreadFactory , <EOL> <NUM_LIT:0> , <EOL> RequestManager , <EOL> DataSourceThread , <EOL> Queue . Queue ) <EOL> try : <EOL> return_code = app . Run ( ) <EOL> except AuthenticationError : <EOL> logger . info ( '<STR_LIT>' ) <EOL> finally : <EOL> exporter . finalize ( ) <EOL> elif not download : <EOL> mapper = Mapper . RegisteredMapper ( kind ) <EOL> try : <EOL> mapper . initialize ( mapper_opts ) <EOL> def KeyRangeGeneratorFactory ( request_manager , progress_queue , <EOL> progress_gen ) : <EOL> return KeyRangeItemGenerator ( request_manager , kind , progress_queue , <EOL> progress_gen , MapperItem ) <EOL> def MapperProgressThreadFactory ( progress_queue , progress_db ) : <EOL> return MapperProgressThread ( kind , <EOL> progress_queue , <EOL> progress_db ) <EOL> app = BulkMapperApp ( arg_dict , <EOL> KeyRangeGeneratorFactory , <EOL> throttle , <EOL> progress_db , <EOL> MapperProgressThreadFactory , <EOL> <NUM_LIT:0> , <EOL> RequestManager , <EOL> DataSourceThread , <EOL> Queue . Queue ) <EOL> try : <EOL> return_code = app . Run ( ) <EOL> except AuthenticationError : <EOL> logger . info ( '<STR_LIT>' ) <EOL> finally : <EOL> mapper . finalize ( ) <EOL> return return_code <EOL> def SetupLogging ( arg_dict ) : <EOL> """<STR_LIT>""" <EOL> format = '<STR_LIT>' <EOL> debug = arg_dict [ '<STR_LIT>' ] <EOL> log_file = arg_dict [ '<STR_LIT>' ] <EOL> logger . setLevel ( logging . DEBUG ) <EOL> logger . propagate = False <EOL> file_handler = logging . FileHandler ( log_file , '<STR_LIT:w>' ) <EOL> file_handler . setLevel ( logging . DEBUG ) <EOL> file_formatter = logging . Formatter ( format ) <EOL> file_handler . setFormatter ( file_formatter ) <EOL> logger . addHandler ( file_handler ) <EOL> console = logging . StreamHandler ( ) <EOL> level = logging . INFO <EOL> if debug : <EOL> level = logging . DEBUG <EOL> console . setLevel ( level ) <EOL> console_format = '<STR_LIT>' <EOL> formatter = logging . Formatter ( console_format ) <EOL> console . setFormatter ( formatter ) <EOL> logger . addHandler ( console ) <EOL> logger . info ( '<STR_LIT>' , log_file ) <EOL> remote_api_throttle . logger . setLevel ( level ) <EOL> remote_api_throttle . logger . addHandler ( file_handler ) <EOL> remote_api_throttle . logger . addHandler ( console ) <EOL> appengine_rpc . logger . setLevel ( logging . WARN ) <EOL> adaptive_thread_pool . logger . setLevel ( logging . DEBUG ) <EOL> adaptive_thread_pool . logger . addHandler ( console ) <EOL> adaptive_thread_pool . logger . addHandler ( file_handler ) <EOL> adaptive_thread_pool . logger . propagate = False <EOL> def Run ( arg_dict ) : <EOL> """<STR_LIT>""" <EOL> arg_dict = ProcessArguments ( arg_dict ) <EOL> SetupLogging ( arg_dict ) <EOL> return _PerformBulkload ( arg_dict ) <EOL> def main ( argv ) : <EOL> """<STR_LIT>""" <EOL> arg_dict = ParseArguments ( argv ) <EOL> errors = [ '<STR_LIT>' % key <EOL> for ( key , value ) in arg_dict . iteritems ( ) <EOL> if value is REQUIRED_OPTION ] <EOL> if errors : <EOL> print >> sys . stderr , '<STR_LIT:\n>' . join ( errors ) <EOL> PrintUsageExit ( <NUM_LIT:1> ) <EOL> SetupLogging ( arg_dict ) <EOL> return _PerformBulkload ( arg_dict ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> sys . exit ( main ( sys . argv ) ) </s>
