Coverage for src/task/examples.py: 0%

1import time 

2 

3from task.custom_task import CustomTask 

4 

5 

6class BasicTask(CustomTask): 

7 """ 

8 A really basic task to show how you should use the CustomTask object 

9 

10 the `_make_subtasks` method accepts both regular functions and nested CustomTasks 

11 You can also convert a function to a CustomTask as done here 

12 """ 

13 

14 def do(self): 

15 print("Working...") 

16 

17 def _make_subtasks(self): 

18 return [CustomTask(lambda: print("Working really hard!!!!"))] 

19 

20 

21counter = 0 

22 

23 

24class RecoverExample(CustomTask): 

25 """ 

26 A simple task that will crash 50% of the time to demonstrate the recovery system 

27 """ 

28 

29 def do(self): 

30 print("Working...") 

31 

32 def _make_subtasks(self): 

33 def raise_an_exc(): 

34 global counter 

35 time.sleep(1) 

36 counter += 1 

37 if counter % 2 == 1: 

38 print("Trying to work...") 

39 raise RuntimeError("Error!!!") 

40 else: 

41 print("I finally could work!!!") 

42 

43 class ErrorTask(CustomTask): 

44 def _make_subtasks(self): 

45 return [raise_an_exc] 

46 

47 return [lambda: print("Working really hard!!!!"), ErrorTask()] 

48 

49 

50class NestedRecoveryExample(CustomTask): 

51 """ 

52 Here we want to demonstrate that the recovery system will also work using nested CustomTasks 

53 """ 

54 

55 def do(self): 

56 print("Working...") 

57 

58 def _make_subtasks(self): 

59 return [CustomTask(lambda: print("Working really hard!!!!")), RecoverExample()] 

60 

61 

62class SimpleMathTask(CustomTask): 

63 """ 

64 In this task, we're basically showing how you can work with task arguments and how to link different subtask together 

65 

66 The way it works is kind of logical, when the task is called, the arguments go inside the `do` call, we then send the 

67 return value to the next subtask 

68 """ 

69 

70 def do(self, x): 

71 return x * 100 

72 

73 def _make_subtasks(self): 

74 return [lambda x: x + 5, lambda x: x / 100] 

75 

76 

77class ComplexMathTask(CustomTask): 

78 """ 

79 This one is just to show that you can link CustomTask to another CustomTask or to a regular function 

80 

81 For this task, the schema would be: 

82 `ComplexMathTask -> SimpleMathTask -> lambda x*2` 

83 

84 The result of the last task will then be returned. 

85 """ 

86 

87 def do(self, x): 

88 return x * 0.5 

89 

90 def _make_subtasks(self): 

91 return [SimpleMathTask, lambda x: x * 2]