google_cloud_tasks_create_http_target_task.py

1import datetime
2import json
3
4from google.cloud import tasks_v2
5from google.protobuf import timestamp_pb2
6
7# Create a client.
8client = tasks_v2.CloudTasksClient()
9
10# TODO(developer): Replace these variables before running the sample.
11project = "my-project-id"
12queue = "my-queue"
13location = "us-central1"
14url = "https://example.com/task_handler"
15payload = "hello"
16
17# Construct the fully qualified queue name.
18parent = client.queue_path(project, location, queue)
19
20# Construct the request body.
21task = {
22    "http_request": {  # Specify the type of request.
23        "http_method": tasks_v2.HttpMethod.POST,
24        "url": url,  # The full url path that the task will be sent to.
25    }
26}
27
28if payload is not None:
29    if isinstance(payload, dict):
30        # Convert dict to JSON string
31        payload = json.dumps(payload)
32        # Specify the content-type of the request.
33        task["http_request"]["headers"] = {"Content-type": "application/json"}
34
35    # The API expects a payload of type bytes.
36    converted_payload = payload.encode()
37
38    # Add the payload to the request.
39    task["http_request"]["body"] = converted_payload
40
41# Use the client to build and send the task.
42response = client.create_task(request={"parent": parent, "task": task})
43
44print(f"Created task {response.name}")