<s> """<STR_LIT>""" <EOL> import copy <EOL> from functools import partial <EOL> from functools import wraps <EOL> import json <EOL> import os <EOL> import time <EOL> import uuid <EOL> from furious . job_utils import decode_callbacks <EOL> from furious . job_utils import encode_callbacks <EOL> from furious . job_utils import get_function_path_and_options <EOL> from furious . job_utils import path_to_reference <EOL> from furious . job_utils import reference_to_path <EOL> from furious import errors <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ASYNC_DEFAULT_QUEUE = '<STR_LIT:default>' <EOL> ASYNC_ENDPOINT = '<STR_LIT>' <EOL> MAX_DEPTH = <NUM_LIT:100> <EOL> MAX_RESTARTS = <NUM_LIT:10> <EOL> DISABLE_RECURSION_CHECK = - <NUM_LIT:1> <EOL> RETRY_SLEEP_SECS = <NUM_LIT:4> <EOL> DEFAULT_RETRY_OPTIONS = { <EOL> '<STR_LIT>' : MAX_RESTARTS <EOL> } <EOL> class Async ( object ) : <EOL> def __init__ ( self , target , args = None , kwargs = None , ** options ) : <EOL> self . _options = { } <EOL> _check_options ( options ) <EOL> self . _update_job ( target , args , kwargs ) <EOL> self . update_options ( ** options ) <EOL> self . _initialize_recursion_depth ( ) <EOL> self . _context_id = self . _get_context_id ( ) <EOL> self . _parent_id = self . _get_parent_id ( ) <EOL> self . _id = self . _get_id ( ) <EOL> self . _execution_context = None <EOL> self . _executing = False <EOL> self . _executed = False <EOL> self . _persistence_engine = None <EOL> self . _result = None <EOL> @ property <EOL> def executed ( self ) : <EOL> return self . _executed <EOL> @ property <EOL> def executing ( self ) : <EOL> return self . _executing <EOL> @ executing . setter <EOL> def executing ( self , executing ) : <EOL> if self . _executed : <EOL> raise errors . AlreadyExecutedError ( <EOL> '<STR_LIT>' ) <EOL> if self . _executing : <EOL> raise errors . AlreadyExecutingError ( <EOL> '<STR_LIT>' ) <EOL> self . _executing = executing <EOL> @ property <EOL> def result ( self ) : <EOL> if not self . executed : <EOL> raise errors . NotExecutedError ( <EOL> '<STR_LIT>' ) <EOL> return self . _result <EOL> @ result . setter <EOL> def result ( self , result ) : <EOL> if not self . _executing : <EOL> raise errors . NotExecutingError ( <EOL> '<STR_LIT>' ) <EOL> self . _result = result <EOL> self . _executing = False <EOL> self . _executed = True <EOL> if self . _options . get ( '<STR_LIT>' ) : <EOL> self . _persist_result ( ) <EOL> def _persist_result ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _prepare_persistence_engine ( ) <EOL> return self . _persistence_engine . store_async_result ( <EOL> self . id , self . result ) <EOL> def _decorate_job ( self ) : <EOL> """<STR_LIT>""" <EOL> function_path = self . job [ <NUM_LIT:0> ] <EOL> func = path_to_reference ( function_path ) <EOL> return func <EOL> @ property <EOL> def function_path ( self ) : <EOL> return self . job [ <NUM_LIT:0> ] <EOL> @ property <EOL> def _function_path ( self ) : <EOL> return self . function_path <EOL> @ property <EOL> def job ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _options [ '<STR_LIT>' ] <EOL> @ property <EOL> def recursion_depth ( self ) : <EOL> """<STR_LIT>""" <EOL> recursion_options = self . _options . get ( '<STR_LIT>' , { } ) <EOL> return recursion_options . get ( '<STR_LIT>' , None ) <EOL> def _initialize_recursion_depth ( self ) : <EOL> """<STR_LIT>""" <EOL> from furious . context import get_current_async <EOL> recursion_options = self . _options . get ( '<STR_LIT>' , { } ) <EOL> current_depth = recursion_options . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> max_depth = recursion_options . get ( '<STR_LIT>' , MAX_DEPTH ) <EOL> try : <EOL> executing_async = get_current_async ( ) <EOL> current_depth = executing_async . recursion_depth <EOL> if max_depth == MAX_DEPTH : <EOL> executing_options = executing_async . get_options ( ) . get ( <EOL> '<STR_LIT>' , { } ) <EOL> max_depth = executing_options . get ( '<STR_LIT>' , max_depth ) <EOL> except errors . NotInContextError : <EOL> pass <EOL> self . update_options ( _recursion = { '<STR_LIT>' : current_depth , <EOL> '<STR_LIT>' : max_depth } ) <EOL> def check_recursion_depth ( self ) : <EOL> """<STR_LIT>""" <EOL> from furious . async import MAX_DEPTH <EOL> recursion_options = self . _options . get ( '<STR_LIT>' , { } ) <EOL> max_depth = recursion_options . get ( '<STR_LIT>' , MAX_DEPTH ) <EOL> if ( max_depth != DISABLE_RECURSION_CHECK and <EOL> self . recursion_depth > max_depth ) : <EOL> raise errors . AsyncRecursionError ( '<STR_LIT>' ) <EOL> def _update_job ( self , target , args , kwargs ) : <EOL> """<STR_LIT>""" <EOL> target_path , options = get_function_path_and_options ( target ) <EOL> assert isinstance ( args , ( tuple , list ) ) or args is None <EOL> assert isinstance ( kwargs , dict ) or kwargs is None <EOL> if options : <EOL> self . update_options ( ** options ) <EOL> self . _options [ '<STR_LIT>' ] = ( target_path , args , kwargs ) <EOL> def set_execution_context ( self , execution_context ) : <EOL> """<STR_LIT>""" <EOL> if self . _execution_context : <EOL> raise errors . AlreadyInContextError <EOL> self . _execution_context = execution_context <EOL> def get_options ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _options <EOL> def update_options ( self , ** options ) : <EOL> """<STR_LIT>""" <EOL> _check_options ( options ) <EOL> if '<STR_LIT>' in options : <EOL> options [ '<STR_LIT>' ] = reference_to_path ( <EOL> options [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT:id>' in options : <EOL> self . _id = options [ '<STR_LIT:id>' ] <EOL> self . _options . update ( options ) <EOL> def get_callbacks ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _options . get ( '<STR_LIT>' , { } ) <EOL> def get_headers ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _options . get ( '<STR_LIT>' , { } ) <EOL> def get_queue ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _options . get ( '<STR_LIT>' , ASYNC_DEFAULT_QUEUE ) <EOL> def get_task_args ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _options . get ( '<STR_LIT>' , { } ) <EOL> def to_task ( self ) : <EOL> """<STR_LIT>""" <EOL> from google . appengine . api . taskqueue import Task <EOL> from google . appengine . api . taskqueue import TaskRetryOptions <EOL> self . _increment_recursion_level ( ) <EOL> self . check_recursion_depth ( ) <EOL> url = "<STR_LIT>" % ( ASYNC_ENDPOINT , self . function_path ) <EOL> kwargs = { <EOL> '<STR_LIT:url>' : url , <EOL> '<STR_LIT>' : self . get_headers ( ) . copy ( ) , <EOL> '<STR_LIT>' : json . dumps ( self . to_dict ( ) ) <EOL> } <EOL> kwargs . update ( copy . deepcopy ( self . get_task_args ( ) ) ) <EOL> retry_options = copy . deepcopy ( DEFAULT_RETRY_OPTIONS ) <EOL> retry_options . update ( kwargs . pop ( '<STR_LIT>' , { } ) ) <EOL> kwargs [ '<STR_LIT>' ] = TaskRetryOptions ( ** retry_options ) <EOL> return Task ( ** kwargs ) <EOL> def start ( self , transactional = False , async = False , rpc = None ) : <EOL> """<STR_LIT>""" <EOL> from google . appengine . api import taskqueue <EOL> task = self . to_task ( ) <EOL> queue = taskqueue . Queue ( name = self . get_queue ( ) ) <EOL> retry_transient = self . _options . get ( '<STR_LIT>' , True ) <EOL> retry_delay = self . _options . get ( '<STR_LIT>' , RETRY_SLEEP_SECS ) <EOL> add = queue . add <EOL> if async : <EOL> add = partial ( queue . add_async , rpc = rpc ) <EOL> try : <EOL> ret = add ( task , transactional = transactional ) <EOL> except taskqueue . TransientError : <EOL> if transactional or not retry_transient : <EOL> raise <EOL> time . sleep ( retry_delay ) <EOL> ret = add ( task , transactional = transactional ) <EOL> except ( taskqueue . TaskAlreadyExistsError , <EOL> taskqueue . TombstonedTaskError ) : <EOL> return <EOL> return ret <EOL> def __deepcopy__ ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> return self <EOL> def to_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> return encode_async_options ( self ) <EOL> @ classmethod <EOL> def from_dict ( cls , async ) : <EOL> """<STR_LIT>""" <EOL> async_options = decode_async_options ( async ) <EOL> target , args , kwargs = async_options . pop ( '<STR_LIT>' ) <EOL> return cls ( target , args , kwargs , ** async_options ) <EOL> def _prepare_persistence_engine ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _persistence_engine : <EOL> return <EOL> persistence_engine = self . _options . get ( '<STR_LIT>' ) <EOL> if persistence_engine : <EOL> self . _persistence_engine = path_to_reference ( persistence_engine ) <EOL> return <EOL> from furious . config import get_default_persistence_engine <EOL> self . _persistence_engine = get_default_persistence_engine ( ) <EOL> def _get_context_id ( self ) : <EOL> """<STR_LIT>""" <EOL> from furious . context import get_current_context <EOL> context_id = self . _options . get ( '<STR_LIT>' ) <EOL> if context_id : <EOL> return context_id <EOL> try : <EOL> context = get_current_context ( ) <EOL> except errors . NotInContextError : <EOL> context = None <EOL> self . update_options ( context_id = None ) <EOL> if context : <EOL> context_id = context . id <EOL> self . update_options ( context_id = context_id ) <EOL> return context_id <EOL> def _get_parent_id ( self ) : <EOL> """<STR_LIT>""" <EOL> parent_id = self . _options . get ( '<STR_LIT>' ) <EOL> if parent_id : <EOL> return parent_id <EOL> from furious . context import get_current_async <EOL> try : <EOL> async = get_current_async ( ) <EOL> except errors . NotInContextError : <EOL> async = None <EOL> if async : <EOL> parent_id = "<STR_LIT::>" . join ( [ async . parent_id . split ( "<STR_LIT::>" ) [ <NUM_LIT:0> ] , async . id ] ) <EOL> else : <EOL> parent_id = self . request_id <EOL> self . update_options ( parent_id = parent_id ) <EOL> return parent_id <EOL> def _get_id ( self ) : <EOL> """<STR_LIT>""" <EOL> id = self . _options . get ( '<STR_LIT:id>' ) <EOL> if id : <EOL> return id <EOL> id = uuid . uuid4 ( ) . hex <EOL> self . update_options ( id = id ) <EOL> return id <EOL> @ property <EOL> def id ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _id <EOL> @ property <EOL> def context_id ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _context_id <EOL> @ property <EOL> def parent_id ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _parent_id <EOL> @ property <EOL> def full_id ( self ) : <EOL> """<STR_LIT>""" <EOL> full_id = "<STR_LIT>" <EOL> if self . parent_id : <EOL> full_id = "<STR_LIT::>" . join ( [ self . parent_id , self . id ] ) <EOL> else : <EOL> full_id = self . id <EOL> if self . context_id : <EOL> full_id = "<STR_LIT:|>" . join ( [ full_id , self . context_id ] ) <EOL> return full_id <EOL> @ property <EOL> def request_id ( self ) : <EOL> return os . environ . get ( '<STR_LIT>' , uuid . uuid4 ( ) . hex ) <EOL> def _increment_recursion_level ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _initialize_recursion_depth ( ) <EOL> recursion_options = self . _options . get ( '<STR_LIT>' , { } ) <EOL> current_depth = recursion_options . get ( '<STR_LIT>' , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> max_depth = recursion_options . get ( '<STR_LIT>' , MAX_DEPTH ) <EOL> self . update_options ( _recursion = { '<STR_LIT>' : current_depth , <EOL> '<STR_LIT>' : max_depth } ) <EOL> @ property <EOL> def context_id ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _context_id : <EOL> self . _context_id = self . _get_context_id ( ) <EOL> self . update_options ( context_id = self . _context_id ) <EOL> return self . _context_id <EOL> def _get_context_id ( self ) : <EOL> """<STR_LIT>""" <EOL> from furious . context import get_current_context <EOL> context_id = self . _options . get ( '<STR_LIT>' ) <EOL> if context_id : <EOL> return context_id <EOL> try : <EOL> context = get_current_context ( ) <EOL> except errors . NotInContextError : <EOL> context = None <EOL> self . update_options ( context_id = None ) <EOL> if context : <EOL> context_id = context . id <EOL> self . update_options ( context_id = context_id ) <EOL> return context_id <EOL> class AsyncResult ( object ) : <EOL> SUCCESS = <NUM_LIT:1> <EOL> ERROR = <NUM_LIT:2> <EOL> ABORT = <NUM_LIT:3> <EOL> def __init__ ( self , payload = None , status = None ) : <EOL> self . payload = payload <EOL> self . status = status <EOL> @ property <EOL> def success ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . status != self . ERROR <EOL> def to_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> return { <EOL> '<STR_LIT:status>' : self . status , <EOL> '<STR_LIT>' : self . _payload_to_dict ( ) <EOL> } <EOL> def _payload_to_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . status != self . ERROR or not self . payload : <EOL> return self . payload <EOL> import traceback <EOL> return { <EOL> "<STR_LIT:error>" : self . payload . error , <EOL> "<STR_LIT:args>" : self . payload . args , <EOL> "<STR_LIT>" : traceback . format_exception ( * self . payload . traceback ) <EOL> } <EOL> def async_from_options ( options ) : <EOL> """<STR_LIT>""" <EOL> _type = options . pop ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> _type = path_to_reference ( _type ) <EOL> return _type . from_dict ( options ) <EOL> def encode_async_options ( async ) : <EOL> """<STR_LIT>""" <EOL> options = copy . deepcopy ( async . _options ) <EOL> options [ '<STR_LIT>' ] = reference_to_path ( async . __class__ ) <EOL> eta = options . get ( '<STR_LIT>' , { } ) . get ( '<STR_LIT>' ) <EOL> if eta : <EOL> options [ '<STR_LIT>' ] [ '<STR_LIT>' ] = time . mktime ( eta . timetuple ( ) ) <EOL> callbacks = async . _options . get ( '<STR_LIT>' ) <EOL> if callbacks : <EOL> options [ '<STR_LIT>' ] = encode_callbacks ( callbacks ) <EOL> if '<STR_LIT>' in options : <EOL> _checker = options . pop ( '<STR_LIT>' ) <EOL> options [ '<STR_LIT>' ] = reference_to_path ( _checker ) <EOL> if '<STR_LIT>' in options : <EOL> _processor = options . pop ( '<STR_LIT>' ) <EOL> options [ '<STR_LIT>' ] = reference_to_path ( _processor ) <EOL> return options <EOL> def decode_async_options ( options ) : <EOL> """<STR_LIT>""" <EOL> async_options = copy . deepcopy ( options ) <EOL> eta = async_options . get ( '<STR_LIT>' , { } ) . get ( '<STR_LIT>' ) <EOL> if eta : <EOL> from datetime import datetime <EOL> async_options [ '<STR_LIT>' ] [ '<STR_LIT>' ] = datetime . fromtimestamp ( eta ) <EOL> callbacks = async_options . get ( '<STR_LIT>' , { } ) <EOL> if callbacks : <EOL> async_options [ '<STR_LIT>' ] = decode_callbacks ( callbacks ) <EOL> if '<STR_LIT>' in options : <EOL> _checker = options [ '<STR_LIT>' ] <EOL> async_options [ '<STR_LIT>' ] = path_to_reference ( _checker ) <EOL> if '<STR_LIT>' in options : <EOL> _processor = options [ '<STR_LIT>' ] <EOL> async_options [ '<STR_LIT>' ] = path_to_reference ( _processor ) <EOL> return async_options <EOL> def defaults ( ** options ) : <EOL> """<STR_LIT>""" <EOL> _check_options ( options ) <EOL> def real_decorator ( function ) : <EOL> function . _async_options = options <EOL> @ wraps ( function ) <EOL> def wrapper ( * args , ** kwargs ) : <EOL> return function ( * args , ** kwargs ) <EOL> return wrapper <EOL> return real_decorator <EOL> def _check_options ( options ) : <EOL> """<STR_LIT>""" <EOL> if not options : <EOL> return <EOL> assert '<STR_LIT>' not in options </s>
