gcloud-aio-taskqueue

import asyncio import os from gcloud_aio_taskqueue import TaskQueueClient from gcloud_aio_taskqueue.types import HttpMethod from gcloud_aio_core.session import Session from gcloud_aio_core.auth import build_credentials async def main(): project_id = os.environ.get("GCP_PROJECT_ID", "your-project-id") queue_name = os.environ.get("GCP_TASK_QUEUE_NAME", "your-queue-name") # Authenticate using GOOGLE_APPLICATION_CREDENTIALS env var or a service account path # Example: GOOGLE_APPLICATION_CREDENTIALS=/path/to/key.json credentials = build_credentials() async with Session(credentials) as session: client = TaskQueueClient( project=project_id, queue=queue_name, session=session, ) try: # Add a task with required parameters (changes from v6.0.0 onwards) task = await client.add_task( url="https://example.com/mytaskendpoint", body=b"Hello World!", payload_type="APPLICATION_OCTET_STREAM", method=HttpMethod.POST, headers={ "Content-Type": "application/octet-stream", "X-Cloud-Task-ETA": "2025-01-01T00:00:00Z" # Example custom header } ) print(f"Task added: {task.name} (ID: {task.id})") # Example: Get a task retrieved_task = await client.get_task(task.name) print(f"Retrieved task: {retrieved_task.name}") # Example: Delete a task await client.delete_task(task.name) print(f"Task {task.name} deleted.") except Exception as e: print(f"An error occurred: {e}") if __name__ == "__main__": asyncio.run(main())