FilesExpand file tree

main

/

hello_patch.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

149 lines (127 loc) · 5.07 KB

main

/

hello_patch.py

Copy path

Top

File metadata and controls

• Code
• Blame

149 lines (127 loc) · 5.07 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

import asyncio

import sys

from concurrent.futures import ThreadPoolExecutor

from dataclasses import dataclass

from datetime import timedelta

from temporalio import activity, exceptions, workflow

from temporalio.client import Client

from temporalio.envconfig import ClientConfig

from temporalio.worker import Worker

# While we could use multiple parameters in the activity, Temporal strongly

# encourages using a single dataclass instead which can have fields added to it

# in a backwards-compatible way.

@dataclass

class ComposeGreetingInput:

greeting: str

name: str

# Basic activity that logs and does string concatenation

@activity.defn

def compose_greeting(input: ComposeGreetingInput) -> str:

activity.logger.info("Running activity with parameter %s" % input)

return f"{input.greeting}, {input.name}!"

# V1 of patch-workflow

@workflow.defn(name="patch-workflow")

class MyWorkflow:

@workflow.run

async def run(self, name: str) -> str:

workflow.logger.info("Running patch-workflow with parameter %s" % name)

greeting = await workflow.execute_activity(

compose_greeting,

ComposeGreetingInput("Hello", name),

start_to_close_timeout=timedelta(seconds=70),

)

return greeting

# V2 of patch-workflow using patched where we have changed newly started

# workflow behavior without changing the behavior of currently running workflows

@workflow.defn(name="patch-workflow")

class MyWorkflowPatched:

@workflow.run

async def run(self, name: str) -> str:

workflow.logger.info("Running patch-workflow with parameter %s" % name)

if workflow.patched("my-patch-v2"):

greeting = await workflow.execute_activity(

compose_greeting,

ComposeGreetingInput("Goodbye", name),

start_to_close_timeout=timedelta(seconds=70),

)

await asyncio.sleep(10)

return greeting

else:

greeting = await workflow.execute_activity(

compose_greeting,

ComposeGreetingInput("Hello", name),

start_to_close_timeout=timedelta(seconds=70),

)

return greeting

# V3 of patch-workflow using deprecate_patch where all the old V1 workflows

# have completed, we no longer need to preserve V1 and now just have V2

@workflow.defn(name="patch-worklow")

class MyWorkflowPatchDeprecated:

@workflow.run

async def run(self, name: str) -> str:

workflow.logger.info("Running patch-workflow with parameter %s" % name)

workflow.deprecate_patch("my-patch-v2")

greeting = await workflow.execute_activity(

compose_greeting,

ComposeGreetingInput("Goodbye", name),

start_to_close_timeout=timedelta(seconds=70),

)

await asyncio.sleep(10)

return greeting

async def main():

# Check Args

if len(sys.argv) > 2:

print(f"Incorrect arguments: {sys.argv[0]} v1|v2|v3")

exit()

if len(sys.argv) <= 1:

print(f"Incorrect arguments: {sys.argv[0]} v1|v2|v3v3")

exit()

if sys.argv[1] != "v1" and sys.argv[1] != "v2" and sys.argv[1] != "v3":

print(f"Incorrect arguments: {sys.argv[0]} v1|v2|v3")

exit()

version = sys.argv[1]

# Uncomment the lines below to see logging output

# import logging

# logging.basicConfig(level=logging.INFO)

# Start client

config = ClientConfig.load_client_connect_config()

config.setdefault("target_host", "localhost:7233")

client = await Client.connect(**config)

# Set workflow_class to the proper class based on version

workflow_class = ""

if version == "v1":

workflow_class = MyWorkflow # type: ignore

elif version == "v2":

workflow_class = MyWorkflowPatched # type: ignore

elif version == "v3":

workflow_class = MyWorkflowPatchDeprecated # type: ignore

else:

print(f"Incorrect arguments: {sys.argv[0]} v1|v2|v3")

exit()

# While the worker is running, use the client to run the workflow and

# print out its result. Check if the workflow is already running and

# if so wait for the existing run to complete. Note, in many production setups,

# the client would be in a completely separate process from the worker.

async with Worker(

client,

task_queue="hello-patch-task-queue",

workflows=[workflow_class], # type: ignore

activities=[compose_greeting],

activity_executor=ThreadPoolExecutor(5),

):

try:

result = await client.execute_workflow(

workflow_class.run, # type: ignore

"World",

id="hello-patch-workflow-id",

task_queue="hello-patch-task-queue",

)

print(f"Result: {result}")

except exceptions.WorkflowAlreadyStartedError:

print(f"Workflow already running")

result = await client.get_workflow_handle(

"hello-patch-workflow-id"

).result()

print(f"Result: {result}")

if __name__ == "__main__":

asyncio.run(main())