FilesExpand file tree

 tmatsuo-patch-2

/

create_http_task.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

138 lines (111 loc) · 4.14 KB

 tmatsuo-patch-2

/

create_http_task.py

Copy path

Top

File metadata and controls

• Code
• Blame

138 lines (111 loc) · 4.14 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

# Copyright 2019 Google LLC All Rights Reserved.

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

from __future__ import print_function

import argparse

def create_http_task(project,

queue,

location,

url,

payload=None,

in_seconds=None,

task_name=None):

# [START cloud_tasks_create_http_task]

"""Create a task for a given queue with an arbitrary payload."""

from google.cloud import tasks_v2

from google.protobuf import timestamp_pb2

import datetime

import json

# Create a client.

client = tasks_v2.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.

# project = 'my-project-id'

# queue = 'my-queue'

# location = 'us-central1'

# url = 'https://example.com/task_handler'

# payload = 'hello' or {'param': 'value'} for application/json

# Construct the fully qualified queue name.

parent = client.queue_path(project, location, queue)

# Construct the request body.

task = {

'http_request': { # Specify the type of request.

'http_method': 'POST',

'url': url # The full url path that the task will be sent to.

}

}

if payload is not None:

if isinstance(payload, dict):

# Convert dict to JSON string

payload = json.dumps(payload)

# specify http content-type to application/json

task['http_request']['headers'] = {'Content-type': 'application/json'}

# The API expects a payload of type bytes.

converted_payload = payload.encode()

# Add the payload to the request.

task['http_request']['body'] = converted_payload

if in_seconds is not None:

# Convert "seconds from now" into an rfc3339 datetime string.

d = datetime.datetime.utcnow() + datetime.timedelta(seconds=in_seconds)

# Create Timestamp protobuf.

timestamp = timestamp_pb2.Timestamp()

timestamp.FromDatetime(d)

# Add the timestamp to the tasks.

task['schedule_time'] = timestamp

if task_name is not None:

# Add the name to tasks.

task['name'] = task_name

# Use the client to build and send the task.

response = client.create_task(parent, task)

print('Created task {}'.format(response.name))

# [END cloud_tasks_create_http_task]

return response

if __name__ == '__main__':

parser = argparse.ArgumentParser(

description=create_http_task.__doc__,

formatter_class=argparse.RawDescriptionHelpFormatter)

parser.add_argument(

'--project',

help='Project of the queue to add the task to.',

required=True,

)

parser.add_argument(

'--queue',

help='ID (short name) of the queue to add the task to.',

required=True,

)

parser.add_argument(

'--location',

help='Location of the queue to add the task to.',

required=True,

)

parser.add_argument(

'--url',

help='The full url path that the request will be sent to.',

required=True,

)

parser.add_argument(

'--payload',

help='Optional payload to attach to the push queue.'

)

parser.add_argument(

'--in_seconds', type=int,

help='The number of seconds from now to schedule task attempt.'

)

parser.add_argument(

'--task_name',

help='Task name of the task to create'

)

args = parser.parse_args()

create_http_task(

args.project, args.queue, args.location, args.url,

args.payload, args.in_seconds, args.task_name)