Cloud Tasks
from google.cloud import tasks_v2beta3LOCATION = 'us-central1'PROJECT_ID = '...'QUEUE_ID = '...'client = tasks_v2beta3.CloudTasksClient()parent = client.queue_path(PROJECT_ID, location=LOCATION, queue=QUEUE_ID)task = { 'http_request': { # Specify the type of request. 'http_method': 'POST', 'url': f"https://{LOCATION}-{PROJECT_ID}.cloudfunctions.net/test_payload" }}response = client.create_task(parent, task)
Cloud Functions
def test_payload(request): name = request.values.get('name') # get query string or POST if request.is_json: name = request.json.get('name') return name
NOTE: Flask Get Request Parameters (GET, POST and JSON).
Query String
from urllib.parse import urlencodedata = {'name': 'Desmond Lua', 'age': 40}query_string = urlencode(data)# name=Desmond+Lua&age=40task = { 'http_request': { # Specify the type of request. 'http_method': 'GET', 'url': f"https://{LOCATION}-{PROJECT_ID}.cloudfunctions.net/test_payload?{query_string}" }}
POST
from urllib.parse import urlencodedata = {'name': 'Desmond Lua', 'age': 40}payload = urlencode(data)# name=Desmond+Lua&age=40task = { 'http_request': { # Specify the type of request. 'http_method': 'POST', 'url': f"https://{LOCATION}-{PROJECT_ID}.cloudfunctions.net/test_payload", 'headers': { 'Content-type': 'application/x-www-form-urlencoded' }, 'body': payload.encode() }}
JSON
import jsondata = {'name': 'Desmond Lua', 'age': 40}payload = json.dumps(data)# name=Desmond+Lua&age=40task = { 'http_request': { # Specify the type of request. 'http_method': 'POST', 'url': f"https://{LOCATION}-{PROJECT_ID}.cloudfunctions.net/test_payload", 'headers': { 'Content-type': 'application/application/json' }, 'body': payload.encode() }}