gcloud-rest-taskqueue

import asyncio import os from gcloud.aio.auth import build_from_env from gcloud.aio.taskqueue import TaskQueueClient async def main(): async with build_from_env() as client_session: # Replace with your actual project ID and desired location project_id = os.environ.get("GCLOUD_PROJECT", "your-gcp-project-id") location = os.environ.get("GCLOUD_TASKQUEUE_LOCATION", "us-east1") # e.g., 'us-central1' queue_name = os.environ.get("GCLOUD_TASKQUEUE_NAME", "my-queue") # e.g., 'my-queue' if not project_id or project_id == "your-gcp-project-id": print("Please set GCLOUD_PROJECT environment variable or replace 'your-gcp-project-id'.") return client = TaskQueueClient( project=project_id, location=location, session=client_session ) task_name = "my-test-task-1" payload = {"data": "hello world", "timestamp": "2023-01-01T12:00:00Z"} try: await client.create_task( project=project_id, location=location, queue_name=queue_name, task_name=task_name, payload=payload ) print(f"Task '{task_name}' created successfully in queue '{queue_name}' in project '{project_id}'.") except Exception as e: print(f"Failed to create task: {e}") if __name__ == '__main__': asyncio.run(main())