<s> import abc <EOL> import collections <EOL> import threading <EOL> from automaton import exceptions as machine_excp <EOL> from automaton import machines <EOL> import fasteners <EOL> import futurist <EOL> from oslo_serialization import jsonutils <EOL> from oslo_utils import reflection <EOL> from oslo_utils import timeutils <EOL> import six <EOL> from taskflow . engines . action_engine import executor <EOL> from taskflow import exceptions as excp <EOL> from taskflow import logging <EOL> from taskflow . types import failure as ft <EOL> from taskflow . utils import schema_utils as su <EOL> WAITING = '<STR_LIT>' <EOL> PENDING = '<STR_LIT>' <EOL> RUNNING = '<STR_LIT>' <EOL> SUCCESS = '<STR_LIT>' <EOL> FAILURE = '<STR_LIT>' <EOL> EVENT = '<STR_LIT>' <EOL> WAITING_STATES = ( WAITING , PENDING ) <EOL> STOP_TIMER_STATES = ( RUNNING , SUCCESS , FAILURE ) <EOL> EXECUTE = '<STR_LIT>' <EOL> REVERT = '<STR_LIT>' <EOL> ACTION_TO_EVENT = { <EOL> EXECUTE : executor . EXECUTED , <EOL> REVERT : executor . REVERTED <EOL> } <EOL> REQUEST_TIMEOUT = <NUM_LIT> <EOL> QUEUE_EXPIRE_TIMEOUT = REQUEST_TIMEOUT <EOL> NOTIFY_PERIOD = <NUM_LIT:5> <EOL> EXPIRES_AFTER = <NUM_LIT> <EOL> NOTIFY = '<STR_LIT>' <EOL> REQUEST = '<STR_LIT>' <EOL> RESPONSE = '<STR_LIT>' <EOL> NO_RESULT = object ( ) <EOL> LOG = logging . getLogger ( __name__ ) <EOL> def make_an_event ( new_state ) : <EOL> """<STR_LIT>""" <EOL> return ( '<STR_LIT>' % new_state ) . lower ( ) <EOL> def build_a_machine ( freeze = True ) : <EOL> """<STR_LIT>""" <EOL> m = machines . FiniteMachine ( ) <EOL> for st in ( WAITING , PENDING , RUNNING ) : <EOL> m . add_state ( st ) <EOL> for st in ( SUCCESS , FAILURE ) : <EOL> m . add_state ( st , terminal = True ) <EOL> m . default_start_state = WAITING <EOL> m . add_transition ( WAITING , PENDING , make_an_event ( PENDING ) ) <EOL> m . add_transition ( WAITING , FAILURE , make_an_event ( FAILURE ) ) <EOL> m . add_transition ( PENDING , RUNNING , make_an_event ( RUNNING ) ) <EOL> m . add_transition ( PENDING , FAILURE , make_an_event ( FAILURE ) ) <EOL> m . add_transition ( RUNNING , FAILURE , make_an_event ( FAILURE ) ) <EOL> m . add_transition ( RUNNING , SUCCESS , make_an_event ( SUCCESS ) ) <EOL> if freeze : <EOL> m . freeze ( ) <EOL> return m <EOL> def failure_to_dict ( failure ) : <EOL> """<STR_LIT>""" <EOL> failure_dict = failure . to_dict ( ) <EOL> try : <EOL> jsonutils . dumps ( failure_dict ) <EOL> return failure_dict <EOL> except ( TypeError , ValueError ) : <EOL> return failure . to_dict ( include_args = False ) <EOL> @ six . add_metaclass ( abc . ABCMeta ) <EOL> class Message ( object ) : <EOL> """<STR_LIT>""" <EOL> def __repr__ ( self ) : <EOL> return ( "<STR_LIT>" <EOL> % ( reflection . get_class_name ( self , fully_qualified = False ) , <EOL> id ( self ) , self . to_dict ( ) ) ) <EOL> @ abc . abstractmethod <EOL> def to_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> class Notify ( Message ) : <EOL> """<STR_LIT>""" <EOL> TYPE = NOTIFY <EOL> RESPONSE_SCHEMA = { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> "<STR_LIT:type>" : "<STR_LIT:string>" , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:string>" , <EOL> } , <EOL> } <EOL> } , <EOL> "<STR_LIT>" : [ "<STR_LIT>" , '<STR_LIT>' ] , <EOL> "<STR_LIT>" : False , <EOL> } <EOL> SENDER_SCHEMA = { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> "<STR_LIT>" : False , <EOL> } <EOL> def __init__ ( self , ** data ) : <EOL> self . _data = data <EOL> @ property <EOL> def topic ( self ) : <EOL> return self . _data . get ( '<STR_LIT>' ) <EOL> @ property <EOL> def tasks ( self ) : <EOL> return self . _data . get ( '<STR_LIT>' ) <EOL> def to_dict ( self ) : <EOL> return self . _data <EOL> @ classmethod <EOL> def validate ( cls , data , response ) : <EOL> if response : <EOL> schema = cls . RESPONSE_SCHEMA <EOL> else : <EOL> schema = cls . SENDER_SCHEMA <EOL> try : <EOL> su . schema_validate ( data , schema ) <EOL> except su . ValidationError as e : <EOL> cls_name = reflection . get_class_name ( cls , fully_qualified = False ) <EOL> if response : <EOL> excp . raise_with_cause ( excp . InvalidFormat , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( cls_name , <EOL> e . message ) , <EOL> cause = e ) <EOL> else : <EOL> excp . raise_with_cause ( excp . InvalidFormat , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( cls_name , <EOL> e . message ) , <EOL> cause = e ) <EOL> _WorkUnit = collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT:action>' , '<STR_LIT>' ] ) <EOL> class Request ( Message ) : <EOL> """<STR_LIT>""" <EOL> TYPE = REQUEST <EOL> SCHEMA = { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:result>' : { } , <EOL> '<STR_LIT>' : { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:type>' : '<STR_LIT:string>' , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:type>' : '<STR_LIT:string>' , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:type>" : "<STR_LIT:string>" , <EOL> } , <EOL> { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> } , <EOL> ] , <EOL> } , <EOL> '<STR_LIT:action>' : { <EOL> "<STR_LIT:type>" : "<STR_LIT:string>" , <EOL> "<STR_LIT>" : list ( six . iterkeys ( ACTION_TO_EVENT ) ) , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> } , <EOL> } , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:action>' ] , <EOL> } <EOL> def __init__ ( self , task , uuid , action , <EOL> arguments , timeout = REQUEST_TIMEOUT , result = NO_RESULT , <EOL> failures = None ) : <EOL> self . _action = action <EOL> self . _event = ACTION_TO_EVENT [ action ] <EOL> self . _arguments = arguments <EOL> self . _result = result <EOL> self . _failures = failures <EOL> self . _watch = timeutils . StopWatch ( duration = timeout ) . start ( ) <EOL> self . _lock = threading . Lock ( ) <EOL> self . _machine = build_a_machine ( ) <EOL> self . _machine . initialize ( ) <EOL> self . task = task <EOL> self . uuid = uuid <EOL> self . created_on = timeutils . now ( ) <EOL> self . future = futurist . Future ( ) <EOL> self . future . atom = task <EOL> @ property <EOL> def current_state ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _machine . current_state <EOL> def set_result ( self , result ) : <EOL> """<STR_LIT>""" <EOL> self . future . set_result ( ( self . _event , result ) ) <EOL> @ property <EOL> def expired ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _machine . current_state in WAITING_STATES : <EOL> return self . _watch . expired ( ) <EOL> return False <EOL> def to_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> request = { <EOL> '<STR_LIT>' : reflection . get_class_name ( self . task ) , <EOL> '<STR_LIT>' : self . task . name , <EOL> '<STR_LIT>' : self . task . version , <EOL> '<STR_LIT:action>' : self . _action , <EOL> '<STR_LIT>' : self . _arguments , <EOL> } <EOL> if self . _result is not NO_RESULT : <EOL> result = self . _result <EOL> if isinstance ( result , ft . Failure ) : <EOL> request [ '<STR_LIT:result>' ] = ( '<STR_LIT>' , failure_to_dict ( result ) ) <EOL> else : <EOL> request [ '<STR_LIT:result>' ] = ( '<STR_LIT:success>' , result ) <EOL> if self . _failures : <EOL> request [ '<STR_LIT>' ] = { } <EOL> for atom_name , failure in six . iteritems ( self . _failures ) : <EOL> request [ '<STR_LIT>' ] [ atom_name ] = failure_to_dict ( failure ) <EOL> return request <EOL> def transition_and_log_error ( self , new_state , logger = None ) : <EOL> """<STR_LIT>""" <EOL> if logger is None : <EOL> logger = LOG <EOL> moved = False <EOL> try : <EOL> moved = self . transition ( new_state ) <EOL> except excp . InvalidState : <EOL> logger . warn ( "<STR_LIT>" , self , <EOL> new_state , exc_info = True ) <EOL> return moved <EOL> @ fasteners . locked <EOL> def transition ( self , new_state ) : <EOL> """<STR_LIT>""" <EOL> old_state = self . _machine . current_state <EOL> if old_state == new_state : <EOL> return False <EOL> try : <EOL> self . _machine . process_event ( make_an_event ( new_state ) ) <EOL> except ( machine_excp . NotFound , machine_excp . InvalidState ) as e : <EOL> raise excp . InvalidState ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( old_state , <EOL> new_state , e ) ) <EOL> else : <EOL> if new_state in STOP_TIMER_STATES : <EOL> self . _watch . stop ( ) <EOL> LOG . debug ( "<STR_LIT>" , self , <EOL> old_state , new_state ) <EOL> return True <EOL> @ classmethod <EOL> def validate ( cls , data ) : <EOL> try : <EOL> su . schema_validate ( data , cls . SCHEMA ) <EOL> except su . ValidationError as e : <EOL> cls_name = reflection . get_class_name ( cls , fully_qualified = False ) <EOL> excp . raise_with_cause ( excp . InvalidFormat , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( cls_name , <EOL> e . message ) , <EOL> cause = e ) <EOL> else : <EOL> failures = [ ] <EOL> if '<STR_LIT>' in data : <EOL> failures . extend ( six . itervalues ( data [ '<STR_LIT>' ] ) ) <EOL> result = data . get ( '<STR_LIT:result>' ) <EOL> if result is not None : <EOL> result_data_type , result_data = result <EOL> if result_data_type == '<STR_LIT>' : <EOL> failures . append ( result_data ) <EOL> for fail_data in failures : <EOL> ft . Failure . validate ( fail_data ) <EOL> @ staticmethod <EOL> def from_dict ( data , task_uuid = None ) : <EOL> """<STR_LIT>""" <EOL> task_cls = data [ '<STR_LIT>' ] <EOL> task_name = data [ '<STR_LIT>' ] <EOL> action = data [ '<STR_LIT:action>' ] <EOL> arguments = data . get ( '<STR_LIT>' , { } ) <EOL> result = data . get ( '<STR_LIT:result>' ) <EOL> failures = data . get ( '<STR_LIT>' ) <EOL> arguments = { <EOL> '<STR_LIT>' : arguments , <EOL> } <EOL> if task_uuid is not None : <EOL> arguments [ '<STR_LIT>' ] = task_uuid <EOL> if result is not None : <EOL> result_data_type , result_data = result <EOL> if result_data_type == '<STR_LIT>' : <EOL> arguments [ '<STR_LIT:result>' ] = ft . Failure . from_dict ( result_data ) <EOL> else : <EOL> arguments [ '<STR_LIT:result>' ] = result_data <EOL> if failures is not None : <EOL> arguments [ '<STR_LIT>' ] = { } <EOL> for task , fail_data in six . iteritems ( failures ) : <EOL> arguments [ '<STR_LIT>' ] [ task ] = ft . Failure . from_dict ( fail_data ) <EOL> return _WorkUnit ( task_cls , task_name , action , arguments ) <EOL> class Response ( Message ) : <EOL> """<STR_LIT>""" <EOL> TYPE = RESPONSE <EOL> SCHEMA = { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:state>' : { <EOL> "<STR_LIT:type>" : "<STR_LIT:string>" , <EOL> "<STR_LIT>" : list ( build_a_machine ( ) . states ) + [ EVENT ] , <EOL> } , <EOL> '<STR_LIT:data>' : { <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> ] , <EOL> } , <EOL> } , <EOL> "<STR_LIT>" : [ "<STR_LIT:state>" , '<STR_LIT:data>' ] , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> "<STR_LIT>" : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:type>' : '<STR_LIT:string>' , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:type>' : '<STR_LIT:object>' , <EOL> } , <EOL> } , <EOL> "<STR_LIT>" : [ "<STR_LIT>" , '<STR_LIT>' ] , <EOL> "<STR_LIT>" : False , <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> "<STR_LIT>" : False , <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:result>" : { } , <EOL> } , <EOL> "<STR_LIT>" : [ "<STR_LIT:result>" ] , <EOL> "<STR_LIT>" : False , <EOL> } , <EOL> } , <EOL> } <EOL> def __init__ ( self , state , ** data ) : <EOL> self . state = state <EOL> self . data = data <EOL> @ classmethod <EOL> def from_dict ( cls , data ) : <EOL> state = data [ '<STR_LIT:state>' ] <EOL> data = data [ '<STR_LIT:data>' ] <EOL> if state == FAILURE and '<STR_LIT:result>' in data : <EOL> data [ '<STR_LIT:result>' ] = ft . Failure . from_dict ( data [ '<STR_LIT:result>' ] ) <EOL> return cls ( state , ** data ) <EOL> def to_dict ( self ) : <EOL> return dict ( state = self . state , data = self . data ) <EOL> @ classmethod <EOL> def validate ( cls , data ) : <EOL> try : <EOL> su . schema_validate ( data , cls . SCHEMA ) <EOL> except su . ValidationError as e : <EOL> cls_name = reflection . get_class_name ( cls , fully_qualified = False ) <EOL> excp . raise_with_cause ( excp . InvalidFormat , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( cls_name , <EOL> e . message ) , <EOL> cause = e ) <EOL> else : <EOL> state = data [ '<STR_LIT:state>' ] <EOL> if state == FAILURE and '<STR_LIT:result>' in data : <EOL> ft . Failure . validate ( data [ '<STR_LIT:result>' ] ) </s>
