Coverage for async-durable-execution/src/async_durable_execution/runner/local/processors/execution.py: 100%

1"""Execution operation processor for handling EXECUTION operation updates.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from ....models import ( 

8 ErrorObject, 

9 Operation, 

10 OperationAction, 

11 OperationUpdate, 

12) 

13from .base import ( 

14 OperationProcessor, 

15) 

16from ...exceptions import ( 

17 InvalidParameterValueException, 

18) 

19 

20 

21VALID_ACTIONS_FOR_EXECUTION = frozenset( 

22 [ 

23 OperationAction.SUCCEED, 

24 OperationAction.FAIL, 

25 ] 

26) 

27 

28 

29class ExecutionProcessor(OperationProcessor): 

30 """Processes EXECUTION operation updates for workflow completion.""" 

31 

32 valid_actions = VALID_ACTIONS_FOR_EXECUTION 

33 

34 @classmethod 

35 def validate(cls, current_state: Operation | None, update: OperationUpdate) -> None: 

36 """Validate EXECUTION operation update.""" 

37 super().validate(current_state, update) 

38 

39 match update.action: 

40 case OperationAction.SUCCEED: 

41 if update.error is not None: 

42 msg_exec_succeed_error: str = ( 

43 "Cannot provide an Error for SUCCEED action." 

44 ) 

45 raise InvalidParameterValueException(msg_exec_succeed_error) 

46 case OperationAction.FAIL: 

47 if update.payload is not None: 

48 msg_exec_fail_payload: str = ( 

49 "Cannot provide a Payload for FAIL action." 

50 ) 

51 raise InvalidParameterValueException(msg_exec_fail_payload) 

52 

53 def process( 

54 self, 

55 update: OperationUpdate, 

56 current_op: Operation | None, # noqa: ARG002 

57 notifier: Any, 

58 execution_arn: str, 

59 ) -> Operation | None: 

60 """Process EXECUTION operation update for workflow completion/failure.""" 

61 match update.action: 

62 case OperationAction.SUCCEED: 

63 notifier.complete_execution(execution_arn, update.payload) 

64 case _: 

65 # intentional. actual service will fail any EXECUTION update that is not SUCCEED. 

66 error = update.error or ErrorObject.from_message( 

67 "There is no error details but EXECUTION checkpoint action is not SUCCEED." 

68 ) 

69 # All EXECUTION failures go through normal fail path 

70 # Timeout/Stop status is set by executor based on the operation that caused it 

71 notifier.fail_execution(execution_arn, error) 

72 # TODO: Svc doesn't actually create checkpoint for EXECUTION. might have to for localrunner though. 

73 return None