forked from GoogleCloudPlatform/python-docs-samples
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcreate_http_task.py
More file actions
131 lines (105 loc) · 3.81 KB
/
create_http_task.py
File metadata and controls
131 lines (105 loc) · 3.81 KB
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
# 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
import datetime
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
# 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'
# 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:
# 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))
return response
# [END cloud_tasks_create_http_task]
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)