Google Cloud Task Queue Client (Asyncio)

7.0.0 · active · verified Wed Apr 15

An asynchronous Python client for interacting with Google Cloud Task Queue. This library is part of the `gcloud-aio` ecosystem, providing idiomatic `asyncio` support for various Google Cloud services. As an independent sub-package, its releases are decoupled from other `gcloud-aio` components, and it's currently at version 7.0.0.

Warnings

breaking The `add_task` method now returns a `Task` object instead of `None`.
Fix: Update your code to expect and handle the returned `Task` object, accessing its `name` or `id` attributes directly. Previously, you would not get a direct return value from `add_task`.
breaking The `payload` parameter for `add_task` was removed and replaced with `body`.
Fix: Replace `payload=my_data` with `body=my_data`. Also, `payload_type` is now a mandatory parameter specifying the content type of the body.
breaking The `method` parameter for `add_task` now requires an `HttpMethod` enum member instead of a string.
Fix: Import `HttpMethod` from `gcloud_aio_taskqueue.types` and use it (e.g., `method=HttpMethod.POST`) instead of string literals (e.g., `method='POST'`).
breaking The `Client` class was renamed to `TaskQueueClient` for better clarity and consistency.
Fix: Update your imports from `from gcloud_aio_taskqueue import Client` to `from gcloud_aio_taskqueue import TaskQueueClient`.
gotcha Always manage `gcloud_aio_core.session.Session` or `aiohttp.ClientSession` as an async context manager (`async with`).
Fix: Ensure your client instantiation and usage is wrapped in an `async with Session(...) as session:` block to properly handle resource cleanup and avoid connection leaks.

Install

pip install gcloud-aio-taskqueue Install latest version

Imports

TaskQueueClient
wrong:
```
from gcloud_aio_taskqueue import Client
```
correct:
```
from gcloud_aio_taskqueue import TaskQueueClient
```
The main client class was renamed from `Client` to `TaskQueueClient` in v7.0.0 to improve clarity and avoid naming conflicts.
Session
```
from gcloud_aio_core.session import Session
```
The core session management class, part of `gcloud-aio-core`.
build_credentials
wrong:
```
from gcloud_aio_core.auth import Authenticator
```
correct:
```
from gcloud_aio_core.auth import build_credentials
```
The recommended way to build credentials; `Authenticator` is an older, less direct pattern.
HttpMethod
```
from gcloud_aio_taskqueue.types import HttpMethod
```
Required enum for specifying HTTP methods when adding tasks since v6.0.0.

Quickstart

This quickstart demonstrates how to initialize the `TaskQueueClient`, authenticate using environment variables or service account paths, and perform common operations like adding, retrieving, and deleting tasks. It highlights the use of `gcloud_aio_core.session.Session` for proper async context management and necessary parameters for `add_task` after recent breaking changes.

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())

view raw JSON →