google_cloud_tasks_create_http_target_task_with_json_payload.py

1import datetime
2import json
3
4from google.cloud import tasks_v2
5from google.protobuf import timestamp_pb2
6
7def create_http_task(
8    project: str,
9    location: str,
10    queue: str,
11    url: str,
12    payload: str = None,
13    in_seconds: int = None,
14) -> tasks_v2.Task:
15    """Create an HTTP Target task with a JSON payload."""
16
17    # Create a client.
18    client = tasks_v2.CloudTasksClient()
19
20    # Construct the fully qualified queue name.
21    parent = client.queue_path(project, location, queue)
22
23    # Construct the request body.
24    task = {
25        "http_request": {  # Specify the type of request.
26            "http_method": tasks_v2.HttpMethod.POST,
27            "url": url,  # The full url path that the task will be sent to.
28        }
29    }
30
31    if payload is not None:
32        if isinstance(payload, dict):
33            # Convert dict to JSON string
34            payload = json.dumps(payload)
35            # Specify the content-type of the request
36            task["http_request"]["headers"] = {"Content-type": "application/json"}
37
38        # The API expects a payload of type bytes.
39        converted_payload = payload.encode()
40
41        # Add the payload to the request.
42        task["http_request"]["body"] = converted_payload
43
44    if in_seconds is not None:
45        # Convert "seconds from now" into an rfc3339 datetime string.
46        d = datetime.datetime.now(tz=datetime.timezone.utc) + datetime.timedelta(
47            seconds=in_seconds
48        )
49
50        # Create Timestamp protobuf.
51        timestamp = timestamp_pb2.Timestamp()
52        timestamp.FromDatetime(d)
53
54        # Add the timestamp to the tasks.
55        task["schedule_time"] = timestamp
56
57    # Use the client to build and send the task.
58    response = client.create_task(request={"parent": parent, "task": task})
59
60    print(f"Created task {response.name}")
61    return response