<s> """<STR_LIT>""" <EOL> import StringIO <EOL> import base64 <EOL> import bisect <EOL> import datetime <EOL> import logging <EOL> import os <EOL> import random <EOL> import string <EOL> import time <EOL> from google . appengine . api import api_base_pb <EOL> from google . appengine . api import apiproxy_stub <EOL> from google . appengine . api import apiproxy_stub_map <EOL> from google . appengine . api import queueinfo <EOL> from google . appengine . api . labs . taskqueue import taskqueue_service_pb <EOL> from google . appengine . runtime import apiproxy_errors <EOL> DEFAULT_RATE = '<STR_LIT>' <EOL> DEFAULT_BUCKET_SIZE = <NUM_LIT:5> <EOL> MAX_ETA_DELTA_DAYS = <NUM_LIT:30> <EOL> admin_console_dummy_tasks = { } <EOL> BUILT_IN_HEADERS = set ( [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> DEFAULT_QUEUE_NAME = '<STR_LIT:default>' <EOL> CRON_QUEUE_NAME = '<STR_LIT>' <EOL> class _DummyTaskStore ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _sorted_by_name = [ ] <EOL> self . _sorted_by_eta = [ ] <EOL> def _InsertTask ( self , task ) : <EOL> """<STR_LIT>""" <EOL> eta = task . eta_usec ( ) <EOL> name = task . task_name ( ) <EOL> bisect . insort_left ( self . _sorted_by_eta , ( eta , name , task ) ) <EOL> bisect . insort_left ( self . _sorted_by_name , ( name , task ) ) <EOL> def Lookup ( self , maximum , name = None , eta = None ) : <EOL> """<STR_LIT>""" <EOL> if eta is None : <EOL> pos = bisect . bisect_left ( self . _sorted_by_name , ( name , ) ) <EOL> tasks = ( x [ <NUM_LIT:1> ] for x in self . _sorted_by_name [ pos : pos + maximum ] ) <EOL> return list ( tasks ) <EOL> if name is None : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> pos = bisect . bisect_left ( self . _sorted_by_eta , ( eta , name ) ) <EOL> tasks = ( x [ <NUM_LIT:2> ] for x in self . _sorted_by_eta [ pos : pos + maximum ] ) <EOL> return list ( tasks ) <EOL> def Count ( self ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . _sorted_by_name ) <EOL> def Oldest ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _sorted_by_eta : <EOL> return self . _sorted_by_eta [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> return None <EOL> def Add ( self , request ) : <EOL> """<STR_LIT>""" <EOL> pos = bisect . bisect_left ( self . _sorted_by_name , ( request . task_name ( ) , ) ) <EOL> if ( pos < len ( self . _sorted_by_name ) and <EOL> self . _sorted_by_name [ pos ] [ <NUM_LIT:0> ] == request . task_name ( ) ) : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . TASK_ALREADY_EXISTS ) <EOL> now = datetime . datetime . utcnow ( ) <EOL> now_sec = time . mktime ( now . timetuple ( ) ) <EOL> task = taskqueue_service_pb . TaskQueueQueryTasksResponse_Task ( ) <EOL> task . set_task_name ( request . task_name ( ) ) <EOL> task . set_eta_usec ( request . eta_usec ( ) ) <EOL> task . set_creation_time_usec ( now_sec * <NUM_LIT> ) <EOL> task . set_url ( request . url ( ) ) <EOL> task . set_method ( request . method ( ) ) <EOL> for keyvalue in task . header_list ( ) : <EOL> header = task . add_header ( ) <EOL> header . set_key ( keyvalue . key ( ) ) <EOL> header . set_value ( keyvalue . value ( ) ) <EOL> if request . has_description ( ) : <EOL> task . set_description ( request . description ( ) ) <EOL> if request . has_body ( ) : <EOL> task . set_body ( request . body ( ) ) <EOL> if request . has_crontimetable ( ) : <EOL> task . mutable_crontimetable ( ) . set_schedule ( <EOL> request . crontimetable ( ) . schedule ( ) ) <EOL> task . mutable_crontimetable ( ) . set_timezone ( <EOL> request . crontimetable ( ) . timezone ( ) ) <EOL> self . _InsertTask ( task ) <EOL> def Delete ( self , name ) : <EOL> """<STR_LIT>""" <EOL> pos = bisect . bisect_left ( self . _sorted_by_name , ( name , ) ) <EOL> if pos >= len ( self . _sorted_by_name ) : <EOL> return taskqueue_service_pb . TaskQueueServiceError . UNKNOWN_TASK <EOL> if self . _sorted_by_name [ pos ] [ <NUM_LIT:1> ] . task_name ( ) != name : <EOL> logging . info ( '<STR_LIT>' , name , <EOL> self . _sorted_by_name [ pos ] [ <NUM_LIT:1> ] . task_name ( ) ) <EOL> return taskqueue_service_pb . TaskQueueServiceError . UNKNOWN_TASK <EOL> old_task = self . _sorted_by_name . pop ( pos ) [ <NUM_LIT:1> ] <EOL> eta = old_task . eta_usec ( ) <EOL> pos = bisect . bisect_left ( self . _sorted_by_eta , ( eta , name , None ) ) <EOL> if self . _sorted_by_eta [ pos ] [ <NUM_LIT:2> ] is not old_task : <EOL> logging . error ( '<STR_LIT>' ) <EOL> return taskqueue_service_pb . TaskQueueServiceError . INTERNAL_ERRROR <EOL> self . _sorted_by_eta . pop ( pos ) <EOL> return taskqueue_service_pb . TaskQueueServiceError . OK <EOL> def Populate ( self , num_tasks ) : <EOL> """<STR_LIT>""" <EOL> now = datetime . datetime . utcnow ( ) <EOL> now_sec = time . mktime ( now . timetuple ( ) ) <EOL> def RandomTask ( ) : <EOL> """<STR_LIT>""" <EOL> task = taskqueue_service_pb . TaskQueueQueryTasksResponse_Task ( ) <EOL> task . set_task_name ( '<STR_LIT>' . join ( random . choice ( string . ascii_lowercase ) <EOL> for x in range ( <NUM_LIT:20> ) ) ) <EOL> task . set_eta_usec ( int ( now_sec * <NUM_LIT> ) + random . randint ( - <NUM_LIT> , <NUM_LIT> ) ) <EOL> task . set_creation_time_usec ( min ( now_sec * <NUM_LIT> , task . eta_usec ( ) ) - <EOL> random . randint ( <NUM_LIT:0> , <NUM_LIT> ) ) <EOL> task . set_url ( random . choice ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> if random . random ( ) < <NUM_LIT> : <EOL> task . set_method ( <EOL> taskqueue_service_pb . TaskQueueQueryTasksResponse_Task . POST ) <EOL> task . set_body ( '<STR_LIT:A>' * <NUM_LIT> ) <EOL> else : <EOL> task . set_method ( <EOL> taskqueue_service_pb . TaskQueueQueryTasksResponse_Task . GET ) <EOL> task . set_retry_count ( max ( <NUM_LIT:0> , random . randint ( - <NUM_LIT:10> , <NUM_LIT:5> ) ) ) <EOL> if random . random ( ) < <NUM_LIT> : <EOL> random_headers = [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> for _ in xrange ( random . randint ( <NUM_LIT:1> , <NUM_LIT:4> ) ) : <EOL> elem = random . randint ( <NUM_LIT:0> , len ( random_headers ) - <NUM_LIT:1> ) <EOL> key , value = random_headers . pop ( elem ) <EOL> header_proto = task . add_header ( ) <EOL> header_proto . set_key ( key ) <EOL> header_proto . set_value ( value ) <EOL> return task <EOL> for _ in range ( num_tasks ) : <EOL> self . _InsertTask ( RandomTask ( ) ) <EOL> def _ParseQueueYaml ( unused_self , root_path ) : <EOL> """<STR_LIT>""" <EOL> if root_path is None : <EOL> return None <EOL> for queueyaml in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> try : <EOL> fh = open ( os . path . join ( root_path , queueyaml ) , '<STR_LIT:r>' ) <EOL> except IOError : <EOL> continue <EOL> try : <EOL> queue_info = queueinfo . LoadSingleQueue ( fh ) <EOL> return queue_info <EOL> finally : <EOL> fh . close ( ) <EOL> return None <EOL> def _CompareTasksByEta ( a , b ) : <EOL> """<STR_LIT>""" <EOL> if a . eta_usec ( ) > b . eta_usec ( ) : <EOL> return <NUM_LIT:1> <EOL> if a . eta_usec ( ) < b . eta_usec ( ) : <EOL> return - <NUM_LIT:1> <EOL> return <NUM_LIT:0> <EOL> def _FormatEta ( eta_usec ) : <EOL> """<STR_LIT>""" <EOL> eta = datetime . datetime . fromtimestamp ( eta_usec / <NUM_LIT> ) <EOL> return eta . strftime ( '<STR_LIT>' ) <EOL> def _EtaDelta ( eta_usec ) : <EOL> """<STR_LIT>""" <EOL> eta = datetime . datetime . fromtimestamp ( eta_usec / <NUM_LIT> ) <EOL> now = datetime . datetime . utcnow ( ) <EOL> if eta > now : <EOL> return str ( eta - now ) + '<STR_LIT>' <EOL> else : <EOL> return str ( now - eta ) + '<STR_LIT>' <EOL> class TaskQueueServiceStub ( apiproxy_stub . APIProxyStub ) : <EOL> """<STR_LIT>""" <EOL> queue_yaml_parser = _ParseQueueYaml <EOL> def __init__ ( self , <EOL> service_name = '<STR_LIT>' , <EOL> root_path = None , <EOL> auto_task_running = False , <EOL> task_retry_seconds = <NUM_LIT:30> , <EOL> _all_queues_valid = False ) : <EOL> """<STR_LIT>""" <EOL> super ( TaskQueueServiceStub , self ) . __init__ ( service_name ) <EOL> self . _taskqueues = { } <EOL> self . _next_task_id = <NUM_LIT:1> <EOL> self . _root_path = root_path <EOL> self . _all_queues_valid = _all_queues_valid <EOL> self . _add_event = None <EOL> self . _auto_task_running = auto_task_running <EOL> self . _task_retry_seconds = task_retry_seconds <EOL> self . _app_queues = { } <EOL> class _QueueDetails ( taskqueue_service_pb . TaskQueueUpdateQueueRequest ) : <EOL> def __init__ ( self , paused = False ) : <EOL> self . paused = paused <EOL> def _ChooseTaskName ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _next_task_id += <NUM_LIT:1> <EOL> return '<STR_LIT>' % ( self . _next_task_id - <NUM_LIT:1> ) <EOL> def _VerifyTaskQueueAddRequest ( self , request ) : <EOL> """<STR_LIT>""" <EOL> if request . eta_usec ( ) < <NUM_LIT:0> : <EOL> return taskqueue_service_pb . TaskQueueServiceError . INVALID_ETA <EOL> eta = datetime . datetime . utcfromtimestamp ( request . eta_usec ( ) / <NUM_LIT> ) <EOL> max_eta = ( datetime . datetime . utcnow ( ) + <EOL> datetime . timedelta ( days = MAX_ETA_DELTA_DAYS ) ) <EOL> if eta > max_eta : <EOL> return taskqueue_service_pb . TaskQueueServiceError . INVALID_ETA <EOL> return taskqueue_service_pb . TaskQueueServiceError . OK <EOL> def _Dynamic_Add ( self , request , response ) : <EOL> bulk_request = taskqueue_service_pb . TaskQueueBulkAddRequest ( ) <EOL> bulk_response = taskqueue_service_pb . TaskQueueBulkAddResponse ( ) <EOL> bulk_request . add_add_request ( ) . CopyFrom ( request ) <EOL> self . _Dynamic_BulkAdd ( bulk_request , bulk_response ) <EOL> assert bulk_response . taskresult_size ( ) == <NUM_LIT:1> <EOL> result = bulk_response . taskresult ( <NUM_LIT:0> ) . result ( ) <EOL> if result != taskqueue_service_pb . TaskQueueServiceError . OK : <EOL> raise apiproxy_errors . ApplicationError ( result ) <EOL> elif bulk_response . taskresult ( <NUM_LIT:0> ) . has_chosen_task_name ( ) : <EOL> response . set_chosen_task_name ( <EOL> bulk_response . taskresult ( <NUM_LIT:0> ) . chosen_task_name ( ) ) <EOL> def _Dynamic_BulkAdd ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> assert request . add_request_size ( ) , '<STR_LIT>' <EOL> app_id = None <EOL> if request . add_request ( <NUM_LIT:0> ) . has_app_id ( ) : <EOL> app_id = request . add_request ( <NUM_LIT:0> ) . app_id ( ) <EOL> if not self . _IsValidQueue ( request . add_request ( <NUM_LIT:0> ) . queue_name ( ) , app_id ) : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . UNKNOWN_QUEUE ) <EOL> error_found = False <EOL> task_results_with_chosen_names = [ ] <EOL> for add_request in request . add_request_list ( ) : <EOL> task_result = response . add_taskresult ( ) <EOL> error = self . _VerifyTaskQueueAddRequest ( add_request ) <EOL> if error == taskqueue_service_pb . TaskQueueServiceError . OK : <EOL> if not add_request . task_name ( ) : <EOL> chosen_name = self . _ChooseTaskName ( ) <EOL> add_request . set_task_name ( chosen_name ) <EOL> task_results_with_chosen_names . append ( task_result ) <EOL> task_result . set_result ( <EOL> taskqueue_service_pb . TaskQueueServiceError . SKIPPED ) <EOL> else : <EOL> error_found = True <EOL> task_result . set_result ( error ) <EOL> if error_found : <EOL> return <EOL> if request . add_request ( <NUM_LIT:0> ) . has_transaction ( ) : <EOL> self . _TransactionalBulkAdd ( request ) <EOL> elif request . add_request ( <NUM_LIT:0> ) . has_app_id ( ) : <EOL> self . _DummyTaskStoreBulkAdd ( request , response ) <EOL> else : <EOL> self . _NonTransactionalBulkAdd ( request , response ) <EOL> for add_request , task_result in zip ( request . add_request_list ( ) , <EOL> response . taskresult_list ( ) ) : <EOL> if ( task_result . result ( ) == <EOL> taskqueue_service_pb . TaskQueueServiceError . SKIPPED ) : <EOL> task_result . set_result ( taskqueue_service_pb . TaskQueueServiceError . OK ) <EOL> if task_result in task_results_with_chosen_names : <EOL> task_result . set_chosen_task_name ( add_request . task_name ( ) ) <EOL> def _TransactionalBulkAdd ( self , request ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> apiproxy_stub_map . MakeSyncCall ( <EOL> '<STR_LIT>' , '<STR_LIT>' , request , api_base_pb . VoidProto ( ) ) <EOL> except apiproxy_errors . ApplicationError , e : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> e . application_error + <EOL> taskqueue_service_pb . TaskQueueServiceError . DATASTORE_ERROR , <EOL> e . error_detail ) <EOL> def _DummyTaskStoreBulkAdd ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> store = self . GetDummyTaskStore ( request . add_request ( <NUM_LIT:0> ) . app_id ( ) , <EOL> request . add_request ( <NUM_LIT:0> ) . queue_name ( ) ) <EOL> for add_request , task_result in zip ( request . add_request_list ( ) , <EOL> response . taskresult_list ( ) ) : <EOL> try : <EOL> store . Add ( add_request ) <EOL> except apiproxy_errors . ApplicationError , e : <EOL> task_result . set_result ( e . application_error ) <EOL> else : <EOL> task_result . set_result ( taskqueue_service_pb . TaskQueueServiceError . OK ) <EOL> def _NonTransactionalBulkAdd ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> existing_tasks = self . _taskqueues . setdefault ( <EOL> request . add_request ( <NUM_LIT:0> ) . queue_name ( ) , [ ] ) <EOL> existing_task_names = set ( task . task_name ( ) for task in existing_tasks ) <EOL> def DefineCallback ( queue_name , task_name ) : <EOL> return lambda : self . _RunTask ( queue_name , task_name ) <EOL> for add_request , task_result in zip ( request . add_request_list ( ) , <EOL> response . taskresult_list ( ) ) : <EOL> if add_request . task_name ( ) in existing_task_names : <EOL> task_result . set_result ( <EOL> taskqueue_service_pb . TaskQueueServiceError . TASK_ALREADY_EXISTS ) <EOL> else : <EOL> existing_tasks . append ( add_request ) <EOL> if self . _add_event and self . _auto_task_running : <EOL> self . _add_event ( <EOL> add_request . eta_usec ( ) / <NUM_LIT> , <EOL> DefineCallback ( add_request . queue_name ( ) , add_request . task_name ( ) ) ) <EOL> existing_tasks . sort ( _CompareTasksByEta ) <EOL> def _IsValidQueue ( self , queue_name , app_id ) : <EOL> """<STR_LIT>""" <EOL> if self . _all_queues_valid : <EOL> return True <EOL> if queue_name == DEFAULT_QUEUE_NAME or queue_name == CRON_QUEUE_NAME : <EOL> return True <EOL> queue_info = self . queue_yaml_parser ( self . _root_path ) <EOL> if queue_info and queue_info . queue : <EOL> for entry in queue_info . queue : <EOL> if entry . name == queue_name : <EOL> return True <EOL> if app_id is not None : <EOL> queues = self . _app_queues . get ( app_id , { } ) <EOL> return queues . get ( queue_name , None ) is not None <EOL> return False <EOL> def _RunTask ( self , queue_name , task_name ) : <EOL> """<STR_LIT>""" <EOL> task_list = self . GetTasks ( queue_name ) <EOL> for task in task_list : <EOL> if task [ '<STR_LIT:name>' ] == task_name : <EOL> break <EOL> else : <EOL> return None <EOL> class FakeConnection ( object ) : <EOL> def __init__ ( self , input_buffer ) : <EOL> self . rfile = StringIO . StringIO ( input_buffer ) <EOL> self . wfile = StringIO . StringIO ( ) <EOL> self . wfile_close = self . wfile . close <EOL> self . wfile . close = self . connection_done <EOL> def connection_done ( myself ) : <EOL> result = myself . wfile . getvalue ( ) <EOL> myself . wfile_close ( ) <EOL> first_line , rest = ( result . split ( '<STR_LIT:\n>' , <NUM_LIT:1> ) + [ '<STR_LIT>' , '<STR_LIT>' ] ) [ : <NUM_LIT:2> ] <EOL> version , code , rest = ( first_line . split ( '<STR_LIT:U+0020>' , <NUM_LIT:2> ) + [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) [ : <NUM_LIT:3> ] <EOL> try : <EOL> code = int ( code ) <EOL> except ValueError : <EOL> code = <NUM_LIT> <EOL> if <NUM_LIT:200> <= int ( code ) <= <NUM_LIT> : <EOL> self . DeleteTask ( queue_name , task_name ) <EOL> return <EOL> logging . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> task_name , queue_name , code , self . _task_retry_seconds ) <EOL> self . _add_event ( <EOL> time . time ( ) + self . _task_retry_seconds , <EOL> lambda : self . _RunTask ( queue_name , task_name ) ) <EOL> def close ( self ) : <EOL> pass <EOL> def makefile ( self , mode , buffsize ) : <EOL> if mode . startswith ( '<STR_LIT:w>' ) : <EOL> return self . wfile <EOL> else : <EOL> return self . rfile <EOL> payload = StringIO . StringIO ( ) <EOL> payload . write ( '<STR_LIT>' % ( task [ '<STR_LIT>' ] , task [ '<STR_LIT:url>' ] ) ) <EOL> for key , value in task [ '<STR_LIT>' ] : <EOL> payload . write ( '<STR_LIT>' % ( key , value ) ) <EOL> payload . write ( '<STR_LIT:\r\n>' ) <EOL> payload . write ( task [ '<STR_LIT:body>' ] ) <EOL> return FakeConnection ( payload . getvalue ( ) ) , ( '<STR_LIT>' , <NUM_LIT> ) <EOL> def GetQueues ( self ) : <EOL> """<STR_LIT>""" <EOL> queues = [ ] <EOL> queue_info = self . queue_yaml_parser ( self . _root_path ) <EOL> has_default = False <EOL> if queue_info and queue_info . queue : <EOL> for entry in queue_info . queue : <EOL> if entry . name == DEFAULT_QUEUE_NAME : <EOL> has_default = True <EOL> queue = { } <EOL> queues . append ( queue ) <EOL> queue [ '<STR_LIT:name>' ] = entry . name <EOL> queue [ '<STR_LIT>' ] = entry . rate <EOL> if entry . bucket_size : <EOL> queue [ '<STR_LIT>' ] = entry . bucket_size <EOL> else : <EOL> queue [ '<STR_LIT>' ] = DEFAULT_BUCKET_SIZE <EOL> tasks = self . _taskqueues . setdefault ( entry . name , [ ] ) <EOL> if tasks : <EOL> queue [ '<STR_LIT>' ] = _FormatEta ( tasks [ <NUM_LIT:0> ] . eta_usec ( ) ) <EOL> queue [ '<STR_LIT>' ] = _EtaDelta ( tasks [ <NUM_LIT:0> ] . eta_usec ( ) ) <EOL> else : <EOL> queue [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> queue [ '<STR_LIT>' ] = len ( tasks ) <EOL> if not has_default : <EOL> queue = { } <EOL> queues . append ( queue ) <EOL> queue [ '<STR_LIT:name>' ] = DEFAULT_QUEUE_NAME <EOL> queue [ '<STR_LIT>' ] = DEFAULT_RATE <EOL> queue [ '<STR_LIT>' ] = DEFAULT_BUCKET_SIZE <EOL> tasks = self . _taskqueues . get ( DEFAULT_QUEUE_NAME , [ ] ) <EOL> if tasks : <EOL> queue [ '<STR_LIT>' ] = _FormatEta ( tasks [ <NUM_LIT:0> ] . eta_usec ( ) ) <EOL> queue [ '<STR_LIT>' ] = _EtaDelta ( tasks [ <NUM_LIT:0> ] . eta_usec ( ) ) <EOL> else : <EOL> queue [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> queue [ '<STR_LIT>' ] = len ( tasks ) <EOL> return queues <EOL> def GetTasks ( self , queue_name ) : <EOL> """<STR_LIT>""" <EOL> tasks = self . _taskqueues . get ( queue_name , [ ] ) <EOL> result_tasks = [ ] <EOL> for task_request in tasks : <EOL> task = { } <EOL> result_tasks . append ( task ) <EOL> task [ '<STR_LIT:name>' ] = task_request . task_name ( ) <EOL> task [ '<STR_LIT>' ] = queue_name <EOL> task [ '<STR_LIT:url>' ] = task_request . url ( ) <EOL> method = task_request . method ( ) <EOL> if method == taskqueue_service_pb . TaskQueueAddRequest . GET : <EOL> task [ '<STR_LIT>' ] = '<STR_LIT:GET>' <EOL> elif method == taskqueue_service_pb . TaskQueueAddRequest . POST : <EOL> task [ '<STR_LIT>' ] = '<STR_LIT:POST>' <EOL> elif method == taskqueue_service_pb . TaskQueueAddRequest . HEAD : <EOL> task [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> elif method == taskqueue_service_pb . TaskQueueAddRequest . PUT : <EOL> task [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> elif method == taskqueue_service_pb . TaskQueueAddRequest . DELETE : <EOL> task [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % method ) <EOL> task [ '<STR_LIT>' ] = _FormatEta ( task_request . eta_usec ( ) ) <EOL> task [ '<STR_LIT>' ] = _EtaDelta ( task_request . eta_usec ( ) ) <EOL> task [ '<STR_LIT:body>' ] = base64 . b64encode ( task_request . body ( ) ) <EOL> headers = [ ( header . key ( ) , header . value ( ) ) <EOL> for header in task_request . header_list ( ) <EOL> if header . key ( ) . lower ( ) not in BUILT_IN_HEADERS ] <EOL> headers . append ( ( '<STR_LIT>' , queue_name ) ) <EOL> headers . append ( ( '<STR_LIT>' , task [ '<STR_LIT:name>' ] ) ) <EOL> headers . append ( ( '<STR_LIT>' , '<STR_LIT:0>' ) ) <EOL> headers . append ( ( '<STR_LIT>' , '<STR_LIT:1>' ) ) <EOL> headers . append ( ( '<STR_LIT>' , len ( task [ '<STR_LIT:body>' ] ) ) ) <EOL> if '<STR_LIT>' not in frozenset ( key . lower ( ) for key , _ in headers ) : <EOL> headers . append ( ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) ) <EOL> task [ '<STR_LIT>' ] = headers <EOL> return result_tasks <EOL> def DeleteTask ( self , queue_name , task_name ) : <EOL> """<STR_LIT>""" <EOL> tasks = self . _taskqueues . get ( queue_name , [ ] ) <EOL> for task in tasks : <EOL> if task . task_name ( ) == task_name : <EOL> tasks . remove ( task ) <EOL> return <EOL> def FlushQueue ( self , queue_name ) : <EOL> """<STR_LIT>""" <EOL> self . _taskqueues [ queue_name ] = [ ] <EOL> def _Dynamic_UpdateQueue ( self , request , unused_response ) : <EOL> """<STR_LIT>""" <EOL> queues = self . _app_queues . setdefault ( request . app_id ( ) , { } ) <EOL> if request . queue_name ( ) in queues and queues [ request . queue_name ( ) ] is None : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . TOMBSTONED_QUEUE ) <EOL> defensive_copy = self . _QueueDetails ( ) <EOL> defensive_copy . CopyFrom ( request ) <EOL> queues [ request . queue_name ( ) ] = defensive_copy <EOL> def _Dynamic_FetchQueues ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> queues = self . _app_queues . get ( request . app_id ( ) , { } ) <EOL> for unused_key , queue in sorted ( queues . items ( ) ) : <EOL> if request . max_rows ( ) == response . queue_size ( ) : <EOL> break <EOL> if queue is None : <EOL> continue <EOL> response_queue = response . add_queue ( ) <EOL> response_queue . set_queue_name ( queue . queue_name ( ) ) <EOL> response_queue . set_bucket_refill_per_second ( <EOL> queue . bucket_refill_per_second ( ) ) <EOL> response_queue . set_bucket_capacity ( queue . bucket_capacity ( ) ) <EOL> response_queue . set_user_specified_rate ( queue . user_specified_rate ( ) ) <EOL> if queue . has_max_concurrent_requests ( ) : <EOL> response_queue . set_max_concurrent_requests ( <EOL> queue . max_concurrent_requests ( ) ) <EOL> response_queue . set_paused ( queue . paused ) <EOL> def _Dynamic_FetchQueueStats ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> for queue in request . queue_name_list ( ) : <EOL> store = self . GetDummyTaskStore ( request . app_id ( ) , queue ) <EOL> stats = response . add_queuestats ( ) <EOL> stats . set_num_tasks ( store . Count ( ) ) <EOL> if stats . num_tasks ( ) == <NUM_LIT:0> : <EOL> stats . set_oldest_eta_usec ( - <NUM_LIT:1> ) <EOL> else : <EOL> stats . set_oldest_eta_usec ( store . Oldest ( ) ) <EOL> if random . randint ( <NUM_LIT:0> , <NUM_LIT:9> ) > <NUM_LIT:0> : <EOL> scanner_info = stats . mutable_scanner_info ( ) <EOL> scanner_info . set_executed_last_minute ( random . randint ( <NUM_LIT:0> , <NUM_LIT:10> ) ) <EOL> scanner_info . set_executed_last_hour ( scanner_info . executed_last_minute ( ) <EOL> + random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> scanner_info . set_sampling_duration_seconds ( random . random ( ) * <NUM_LIT> ) <EOL> scanner_info . set_requests_in_flight ( random . randint ( <NUM_LIT:0> , <NUM_LIT:10> ) ) <EOL> def GetDummyTaskStore ( self , app_id , queue_name ) : <EOL> """<STR_LIT>""" <EOL> task_store_key = ( app_id , queue_name ) <EOL> if task_store_key not in admin_console_dummy_tasks : <EOL> store = _DummyTaskStore ( ) <EOL> if not self . _all_queues_valid and queue_name != CRON_QUEUE_NAME : <EOL> store . Populate ( random . randint ( <NUM_LIT:10> , <NUM_LIT:100> ) ) <EOL> admin_console_dummy_tasks [ task_store_key ] = store <EOL> else : <EOL> store = admin_console_dummy_tasks [ task_store_key ] <EOL> return store <EOL> def _Dynamic_QueryTasks ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> store = self . GetDummyTaskStore ( request . app_id ( ) , request . queue_name ( ) ) <EOL> if request . has_start_eta_usec ( ) : <EOL> tasks = store . Lookup ( request . max_rows ( ) , name = request . start_task_name ( ) , <EOL> eta = request . start_eta_usec ( ) ) <EOL> else : <EOL> tasks = store . Lookup ( request . max_rows ( ) , name = request . start_task_name ( ) ) <EOL> for task in tasks : <EOL> response . add_task ( ) . MergeFrom ( task ) <EOL> def _Dynamic_Delete ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> task_store_key = ( request . app_id ( ) , request . queue_name ( ) ) <EOL> if task_store_key not in admin_console_dummy_tasks : <EOL> for _ in request . task_name_list ( ) : <EOL> response . add_result ( <EOL> taskqueue_service_pb . TaskQueueServiceError . UNKNOWN_QUEUE ) <EOL> return <EOL> store = admin_console_dummy_tasks [ task_store_key ] <EOL> for taskname in request . task_name_list ( ) : <EOL> if random . random ( ) <= <NUM_LIT> : <EOL> response . add_result ( <EOL> taskqueue_service_pb . TaskQueueServiceError . TRANSIENT_ERROR ) <EOL> else : <EOL> response . add_result ( store . Delete ( taskname ) ) <EOL> def _Dynamic_ForceRun ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> if random . random ( ) <= <NUM_LIT> : <EOL> response . set_result ( <EOL> taskqueue_service_pb . TaskQueueServiceError . TRANSIENT_ERROR ) <EOL> elif random . random ( ) <= <NUM_LIT> : <EOL> response . set_result ( <EOL> taskqueue_service_pb . TaskQueueServiceError . INTERNAL_ERROR ) <EOL> else : <EOL> response . set_result ( <EOL> taskqueue_service_pb . TaskQueueServiceError . OK ) <EOL> def _Dynamic_DeleteQueue ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> if not request . queue_name ( ) : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . INVALID_QUEUE_NAME ) <EOL> queues = self . _app_queues . get ( request . app_id ( ) , { } ) <EOL> if request . queue_name ( ) not in queues : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . UNKNOWN_QUEUE ) <EOL> elif queues [ request . queue_name ( ) ] is None : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . TOMBSTONED_QUEUE ) <EOL> queues [ request . queue_name ( ) ] = None <EOL> def _Dynamic_PauseQueue ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> if not request . queue_name ( ) : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . INVALID_QUEUE_NAME ) <EOL> queues = self . _app_queues . get ( request . app_id ( ) , { } ) <EOL> if request . queue_name ( ) != DEFAULT_QUEUE_NAME : <EOL> if request . queue_name ( ) not in queues : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . UNKNOWN_QUEUE ) <EOL> elif queues [ request . queue_name ( ) ] is None : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . TOMBSTONED_QUEUE ) <EOL> queues [ request . queue_name ( ) ] . paused = request . pause ( ) <EOL> def _Dynamic_PurgeQueue ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> if not request . queue_name ( ) : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . INVALID_QUEUE_NAME ) <EOL> if request . has_app_id ( ) : <EOL> queues = self . _app_queues . get ( request . app_id ( ) , { } ) <EOL> if request . queue_name ( ) != DEFAULT_QUEUE_NAME : <EOL> if request . queue_name ( ) not in queues : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . UNKNOWN_QUEUE ) <EOL> elif queues [ request . queue_name ( ) ] is None : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . TOMBSTONED_QUEUE ) <EOL> store = self . GetDummyTaskStore ( request . app_id ( ) , request . queue_name ( ) ) <EOL> for task in store . Lookup ( store . Count ( ) ) : <EOL> store . Delete ( task . task_name ( ) ) <EOL> elif ( not self . _IsValidQueue ( request . queue_name ( ) , None ) <EOL> and not request . queue_name ( ) in self . _taskqueues ) : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . UNKNOWN_QUEUE ) <EOL> self . FlushQueue ( request . queue_name ( ) ) <EOL> def _Dynamic_DeleteGroup ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> queues = self . _app_queues . get ( request . app_id ( ) , { } ) <EOL> for queue in queues . iterkeys ( ) : <EOL> store = self . GetDummyTaskStore ( request . app_id ( ) , queue ) <EOL> for task in store . Lookup ( store . Count ( ) ) : <EOL> store . Delete ( task . task_name ( ) ) <EOL> self . FlushQueue ( queue ) <EOL> self . _app_queues [ request . app_id ( ) ] = { } <EOL> def _Dynamic_UpdateStorageLimit ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> if request . limit ( ) < <NUM_LIT:0> or request . limit ( ) > <NUM_LIT:1000> * ( <NUM_LIT> ** <NUM_LIT:4> ) : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . INVALID_REQUEST ) <EOL> response . set_new_limit ( request . limit ( ) ) </s>
