FilesExpand file tree

main

/

workflows.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

189 lines (154 loc) · 6.76 KB

main

/

workflows.py

Copy path

Top

File metadata and controls

• Code
• Blame

189 lines (154 loc) · 6.76 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

"""Workflow definitions for the worker versioning sample."""

from datetime import timedelta

from temporalio import common, workflow

with workflow.unsafe.imports_passed_through():

from worker_versioning.activities import (

IncompatibleActivityInput,

some_activity,

some_incompatible_activity,

)

@workflow.defn(

name="AutoUpgrading", versioning_behavior=common.VersioningBehavior.AUTO_UPGRADE

)

class AutoUpgradingWorkflowV1:

"""AutoUpgradingWorkflowV1 will automatically move to the latest worker version. We'll be making

changes to it, which must be replay safe.

Note that generally you won't want or need to include a version number in your workflow name if

you're using the worker versioning feature. This sample does it to illustrate changes to the

same code over time - but really what we're demonstrating here is the evolution of what would

have been one workflow definition.

"""

def __init__(self) -> None:

self.signals: list[str] = []

@workflow.run

async def run(self) -> None:

workflow.logger.info(

"Changing workflow v1 started.", extra={"StartTime": workflow.now()}

)

# This workflow will listen for signals from our starter, and upon each signal either run

# an activity, or conclude execution.

while True:

await workflow.wait_condition(lambda: len(self.signals) > 0)

signal = self.signals.pop(0)

if signal == "do-activity":

workflow.logger.info("Changing workflow v1 running activity")

await workflow.execute_activity(

some_activity, "v1", start_to_close_timeout=timedelta(seconds=10)

)

else:

workflow.logger.info("Concluding workflow v1")

return

@workflow.signal

async def do_next_signal(self, signal: str) -> None:

"""Signal to perform next action."""

self.signals.append(signal)

@workflow.defn(

name="AutoUpgrading", versioning_behavior=common.VersioningBehavior.AUTO_UPGRADE

)

class AutoUpgradingWorkflowV1b:

"""AutoUpgradingWorkflowV1b represents us having made *compatible* changes to

AutoUpgradingWorkflowV1.

The compatible changes we've made are:

- Altering the log lines

- Using the workflow.patched API to properly introduce branching behavior while maintaining

compatibility

"""

def __init__(self) -> None:

self.signals: list[str] = []

@workflow.run

async def run(self) -> None:

workflow.logger.info(

"Changing workflow v1b started.", extra={"StartTime": workflow.now()}

)

# This workflow will listen for signals from our starter, and upon each signal either run

# an activity, or conclude execution.

while True:

await workflow.wait_condition(lambda: len(self.signals) > 0)

signal = self.signals.pop(0)

if signal == "do-activity":

workflow.logger.info("Changing workflow v1b running activity")

if workflow.patched("DifferentActivity"):

await workflow.execute_activity(

some_incompatible_activity,

IncompatibleActivityInput(called_by="v1b", more_data="hello!"),

start_to_close_timeout=timedelta(seconds=10),

)

else:

# Note it is a valid compatible change to alter the input to an activity.

# However, because we're using the patched API, this branch will never be

# taken.

await workflow.execute_activity(

some_activity,

"v1b",

start_to_close_timeout=timedelta(seconds=10),

)

else:

workflow.logger.info("Concluding workflow v1b")

break

@workflow.signal

async def do_next_signal(self, signal: str) -> None:

"""Signal to perform next action."""

self.signals.append(signal)

@workflow.defn(name="Pinned", versioning_behavior=common.VersioningBehavior.PINNED)

class PinnedWorkflowV1:

"""PinnedWorkflowV1 demonstrates a workflow that likely has a short lifetime, and we want to always

stay pinned to the same version it began on.

Note that generally you won't want or need to include a version number in your workflow name if

you're using the worker versioning feature. This sample does it to illustrate changes to the

same code over time - but really what we're demonstrating here is the evolution of what would

have been one workflow definition.

"""

def __init__(self) -> None:

self.signals: list[str] = []

@workflow.run

async def run(self) -> None:

workflow.logger.info(

"Pinned Workflow v1 started.", extra={"StartTime": workflow.now()}

)

while True:

await workflow.wait_condition(lambda: len(self.signals) > 0)

signal = self.signals.pop(0)

if signal == "conclude":

break

await workflow.execute_activity(

some_activity,

"Pinned-v1",

start_to_close_timeout=timedelta(seconds=10),

)

@workflow.signal

async def do_next_signal(self, signal: str) -> None:

"""Signal to perform next action."""

self.signals.append(signal)

@workflow.defn(name="Pinned", versioning_behavior=common.VersioningBehavior.PINNED)

class PinnedWorkflowV2:

"""PinnedWorkflowV2 has changes that would make it incompatible with v1, and aren't protected by

a patch.

"""

def __init__(self) -> None:

self.signals: list[str] = []

@workflow.run

async def run(self) -> None:

workflow.logger.info(

"Pinned Workflow v2 started.", extra={"StartTime": workflow.now()}

)

# Here we call an activity where we didn't before, which is an incompatible change.

await workflow.execute_activity(

some_activity,

"Pinned-v2",

start_to_close_timeout=timedelta(seconds=10),

)

while True:

await workflow.wait_condition(lambda: len(self.signals) > 0)

signal = self.signals.pop(0)

if signal == "conclude":

break

# We've also changed the activity type here, another incompatible change

await workflow.execute_activity(

some_incompatible_activity,

IncompatibleActivityInput(called_by="Pinned-v2", more_data="hi"),

start_to_close_timeout=timedelta(seconds=10),

)

@workflow.signal

async def do_next_signal(self, signal: str) -> None:

"""Signal to perform next action."""

self.signals.append(signal)