o bi @sddlZddlZddlmZmZddlmZerddlmZdZ dZ dZ Gdd d Z d ed dfd d Z d ed ee fddZde d efddZde d efddZdS)N) TYPE_CHECKINGList) DataContext)StreamingExecutorexecution_callbacksRAY_DATA_EXECUTION_CALLBACKS_env_callbacks_initializedc@s@eZdZdZdddZdddZddd Zddd efd d Zd S)ExecutionCallbackz(Callback interface for execution events.executorrcCdS)z+Called before the Dataset execution starts.Nselfr r r c/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/_internal/execution/execution_callback.pybefore_execution_startsz)ExecutionCallback.before_execution_startscCr )z2Called at each step of the Dataset execution loop.Nr r r r ron_execution_steprz#ExecutionCallback.on_execution_stepcCr )z,Called after the Dataset execution succeeds.Nr r r r rafter_execution_succeedsrz*ExecutionCallback.after_execution_succeedserrorcCr )z)Called after the Dataset execution fails.Nr )rr rr r rafter_execution_failsrz'ExecutionCallback.after_execution_failsN)r r) __name__ __module__ __qualname____doc__rrr Exceptionrr r r rr s    r contextreturnc Cstjtd}|s dS|dD]>}|}|sqz|dd\}}t|}t ||}|}t ||Wqt t t fyN}z t d|d|d}~wwdS)zKInitialize callbacks from environment variable and add them to the context.N,.z Failed to import callback from 'z': )osenvirongetEXECUTION_CALLBACKS_ENV_VARsplitstriprsplit importlib import_modulegetattradd_execution_callback ImportErrorAttributeError ValueError) r callbacks_str callback_path module_path class_namemodule callback_clscallbacker r r_initialize_env_callbacks#s$  r7cCs,|tdst||td|tgS)z0Get all ExecutionCallbacks from the DataContext.FT) get_configENV_CALLBACKS_INITIALIZED_KEYr7 set_configEXECUTION_CALLBACKS_CONFIG_KEY)rr r rget_execution_callbacks8s   r<r5cC&|tg}|||t|dS)z,Add an ExecutionCallback to the DataContext.N)r8r;appendr:r5rrr r rr+B  r+cCr=)z1Remove an ExecutionCallback from the DataContext.N)r8r;remover:r?r r rremove_execution_callbackIr@rB)r(r!typingrrray.data.contextr/ray.data._internal.execution.streaming_executorrr;r$r9r r7r<r+rBr r r rs