<s> import abc <EOL> import copy <EOL> from oslo_utils import reflection <EOL> import six <EOL> from six . moves import map as compat_map <EOL> from six . moves import reduce as compat_reduce <EOL> from taskflow import atom <EOL> from taskflow import logging <EOL> from taskflow . types import notifier <EOL> from taskflow . utils import misc <EOL> LOG = logging . getLogger ( __name__ ) <EOL> REVERT_RESULT = '<STR_LIT:result>' <EOL> REVERT_FLOW_FAILURES = '<STR_LIT>' <EOL> EVENT_UPDATE_PROGRESS = '<STR_LIT>' <EOL> @ six . add_metaclass ( abc . ABCMeta ) <EOL> class Task ( atom . Atom ) : <EOL> """<STR_LIT>""" <EOL> TASK_EVENTS = ( EVENT_UPDATE_PROGRESS , ) <EOL> def __init__ ( self , name = None , provides = None , requires = None , <EOL> auto_extract = True , rebind = None , inject = None , <EOL> ignore_list = None , revert_rebind = None , revert_requires = None ) : <EOL> if name is None : <EOL> name = reflection . get_class_name ( self ) <EOL> super ( Task , self ) . __init__ ( name , provides = provides , requires = requires , <EOL> auto_extract = auto_extract , rebind = rebind , <EOL> inject = inject , revert_rebind = revert_rebind , <EOL> revert_requires = revert_requires ) <EOL> self . _notifier = notifier . RestrictedNotifier ( self . TASK_EVENTS ) <EOL> @ property <EOL> def notifier ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _notifier <EOL> def copy ( self , retain_listeners = True ) : <EOL> """<STR_LIT>""" <EOL> c = copy . copy ( self ) <EOL> c . _notifier = self . _notifier . copy ( ) <EOL> if not retain_listeners : <EOL> c . _notifier . reset ( ) <EOL> return c <EOL> def update_progress ( self , progress ) : <EOL> """<STR_LIT>""" <EOL> def on_clamped ( ) : <EOL> LOG . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" , progress ) <EOL> cleaned_progress = misc . clamp ( progress , <NUM_LIT:0.0> , <NUM_LIT:1.0> , <EOL> on_clamped = on_clamped ) <EOL> self . _notifier . notify ( EVENT_UPDATE_PROGRESS , <EOL> { '<STR_LIT>' : cleaned_progress } ) <EOL> class FunctorTask ( Task ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , execute , name = None , provides = None , <EOL> requires = None , auto_extract = True , rebind = None , revert = None , <EOL> version = None , inject = None ) : <EOL> if not six . callable ( execute ) : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if revert is not None : <EOL> if not six . callable ( revert ) : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if name is None : <EOL> name = reflection . get_callable_name ( execute ) <EOL> super ( FunctorTask , self ) . __init__ ( name , provides = provides , <EOL> inject = inject ) <EOL> self . _execute = execute <EOL> self . _revert = revert <EOL> if version is not None : <EOL> self . version = version <EOL> mapping = self . _build_arg_mapping ( execute , requires , rebind , <EOL> auto_extract ) <EOL> self . rebind , exec_requires , self . optional = mapping <EOL> if revert : <EOL> revert_mapping = self . _build_arg_mapping ( revert , requires , rebind , <EOL> auto_extract ) <EOL> else : <EOL> revert_mapping = ( self . rebind , exec_requires , self . optional ) <EOL> ( self . revert_rebind , revert_requires , <EOL> self . revert_optional ) = revert_mapping <EOL> self . requires = exec_requires . union ( revert_requires ) <EOL> def execute ( self , * args , ** kwargs ) : <EOL> return self . _execute ( * args , ** kwargs ) <EOL> def revert ( self , * args , ** kwargs ) : <EOL> if self . _revert : <EOL> return self . _revert ( * args , ** kwargs ) <EOL> else : <EOL> return None <EOL> class ReduceFunctorTask ( Task ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , functor , requires , name = None , provides = None , <EOL> auto_extract = True , rebind = None , inject = None ) : <EOL> if not six . callable ( functor ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> f_args = reflection . get_callable_args ( functor ) <EOL> if len ( f_args ) != <NUM_LIT:2> : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" % len ( f_args ) ) <EOL> if not misc . is_iterable ( requires ) : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" % type ( requires ) ) <EOL> if len ( requires ) < <NUM_LIT:2> : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" % len ( requires ) ) <EOL> if name is None : <EOL> name = reflection . get_callable_name ( functor ) <EOL> super ( ReduceFunctorTask , self ) . __init__ ( name = name , <EOL> provides = provides , <EOL> inject = inject , <EOL> requires = requires , <EOL> rebind = rebind , <EOL> auto_extract = auto_extract ) <EOL> self . _functor = functor <EOL> def execute ( self , * args , ** kwargs ) : <EOL> l = [ kwargs [ r ] for r in self . requires ] <EOL> return compat_reduce ( self . _functor , l ) <EOL> class MapFunctorTask ( Task ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , functor , requires , name = None , provides = None , <EOL> auto_extract = True , rebind = None , inject = None ) : <EOL> if not six . callable ( functor ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> f_args = reflection . get_callable_args ( functor ) <EOL> if len ( f_args ) != <NUM_LIT:1> : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" % len ( f_args ) ) <EOL> if not misc . is_iterable ( requires ) : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" % type ( requires ) ) <EOL> if name is None : <EOL> name = reflection . get_callable_name ( functor ) <EOL> super ( MapFunctorTask , self ) . __init__ ( name = name , provides = provides , <EOL> inject = inject , requires = requires , <EOL> rebind = rebind , <EOL> auto_extract = auto_extract ) <EOL> self . _functor = functor <EOL> def execute ( self , * args , ** kwargs ) : <EOL> l = [ kwargs [ r ] for r in self . requires ] <EOL> return list ( compat_map ( self . _functor , l ) ) </s>
