Google Cloud Tasks Python Client

Warnings

• gotcha Authentication is crucial. For local development, set the `GOOGLE_APPLICATION_CREDENTIALS` environment variable to point to a service account key file. For deployed applications, ensure the service account running your code has the `Cloud Tasks Enqueuer` role (or `cloudtasks.tasks.create` permission) for the queue, and if targeting authenticated endpoints (like Cloud Run), the task's OIDC token's service account must have permissions to invoke the target.
  Fix: Set `GOOGLE_APPLICATION_CREDENTIALS` locally. Grant `roles/cloudtasks.enqueuer` to the caller for the queue, and appropriate invocation roles (e.g., `roles/run.invoker`) to the service account specified in `oidc_token.service_account_email` if using authenticated HTTP targets.
• gotcha Cloud Tasks aims for exactly-once delivery but can occasionally execute tasks more than once due to system trade-offs. Your task handlers should be idempotent to safely handle duplicate executions without unintended side effects.
  Fix: Design your task handlers to be idempotent. This means the outcome of processing a task multiple times is the same as processing it once. Techniques include using unique transaction IDs or checking state before applying changes.
• gotcha Overwhelming your target service or improperly configuring queue rate limits can lead to backlogs and HTTP 429 (Too Many Requests) or 503 (Service Unavailable) errors. Cloud Tasks will implement backoff, but sustained issues require intervention.
  Fix: Configure appropriate rate limits and concurrency settings on your Cloud Tasks queues. Ensure your target services (e.g., Cloud Run, App Engine) are scaled sufficiently to handle the expected task dispatch rate. Implement client-side rate limiting or exponential backoff when enqueueing tasks if you frequently hit `ResourceExhausted` errors.
• gotcha When sending a payload with an HTTP task, the body must be explicitly encoded to bytes (e.g., `json.dumps(payload).encode('utf-8')`). Failing to encode will result in type errors.
  Fix: Always encode the `body` field of an `HttpRequest` object to bytes using `.encode('utf-8')` (or similar) before sending.
• breaking The client library requires Python 3.9 or newer. Older Python versions (3.8 and below) are unsupported.
  Fix: Upgrade your Python environment to 3.9 or higher. For older Python versions, you might need to use an older version of the `google-cloud-tasks` library, but this is not recommended.
• gotcha Default logging for `google-cloud-tasks` (and other `google` prefixed libraries) does not propagate to the root logger by default. If you expect to see logs via your application's root logger, you must explicitly enable propagation.
  Fix: Add `import logging; logging.getLogger("google").propagate = True` to your application's initialization code if you want Google library logs to be handled by the root logger.

Install

• pip install google-cloud-tasks Install latest version

Imports

• CloudTasksClient

  from google.cloud import tasks_v2

  The primary client for interacting with Cloud Tasks. The `_v2` suffix indicates the API version.
• Task

  from google.cloud.tasks_v2.types import Task

  Represents a unit of scheduled work.
• HttpRequest

  from google.cloud.tasks_v2.types import HttpRequest

  Used to configure an HTTP target for a task.
• HttpMethod

  from google.cloud.tasks_v2.types import HttpMethod

  Enum for specifying the HTTP method of a task's request.
• OidcToken

  from google.cloud.tasks_v2.types import OidcToken

  Used for authenticating tasks to Cloud Run or other OIDC-enabled services.
• timestamp_pb2

  from google.protobuf import timestamp_pb2

  Required for setting `schedule_time` for tasks using protobuf timestamps.

Quickstart

This quickstart demonstrates how to create an HTTP task with OIDC authentication, suitable for targeting services like Cloud Run or Cloud Functions that require identity verification. It includes setting a payload and an optional schedule time. Ensure your Google Cloud environment is configured for Application Default Credentials (ADC) or explicitly set `GOOGLE_APPLICATION_CREDENTIALS` for local development.

import os
import json
import datetime
from google.cloud import tasks_v2
from google.cloud.tasks_v2.types import Task, HttpRequest, HttpMethod
from google.protobuf import timestamp_pb2

def create_http_task_with_auth(
    project_id: str,
    location_id: str,
    queue_id: str,
    url: str,
    service_account_email: str,
    payload: dict = None,
    schedule_time_seconds: int = None,
) -> Task:
    """Create an HTTP task with OIDC authentication.

    Args:
        project_id: The GCP project ID.
        location_id: The location of the queue (e.g., 'us-central1').
        queue_id: The ID of the Cloud Tasks queue.
        url: The full URL of the HTTP target (e.g., a Cloud Run service).
        service_account_email: The email of the service account for OIDC token.
        payload: (Optional) Dictionary payload to send with the task.
        schedule_time_seconds: (Optional) Delay in seconds before task execution.

    Returns:
        The created Task object.
    """
    client = tasks_v2.CloudTasksClient()

    # Construct the queue path.
    parent = client.queue_path(project_id, location_id, queue_id)

    # Construct the HTTP request for the task.
    http_request = HttpRequest(
        http_method=HttpMethod.POST,
        url=url,
        oidc_token={'service_account_email': service_account_email}
    )

    if payload:
        http_request.headers['Content-Type'] = 'application/json'
        http_request.body = json.dumps(payload).encode('utf-8')

    # Construct the task.
    task = Task(http_request=http_request)

    if schedule_time_seconds:
        # Schedule the task for a future time.
        future_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=schedule_time_seconds)
        timestamp = timestamp_pb2.Timestamp()
        timestamp.FromDatetime(future_time)
        task.schedule_time = timestamp

    # Send the task creation request.
    response = client.create_task(parent=parent, task=task)
    print(f"Created task: {response.name}")
    return response

# --- Example Usage ---
if __name__ == "__main__":
    # Set environment variables for authentication (e.g., GOOGLE_APPLICATION_CREDENTIALS)
    # and for the task details.
    project = os.environ.get('GCP_PROJECT_ID', 'your-gcp-project-id')
    location = os.environ.get('GCP_LOCATION_ID', 'us-central1')
    queue = os.environ.get('GCP_TASK_QUEUE_ID', 'my-http-queue')
    target_url = os.environ.get('TASK_TARGET_URL', 'https://your-cloud-run-service-url.run.app/task-handler')
    sa_email = os.environ.get('TASK_SERVICE_ACCOUNT_EMAIL', 'your-service-account@your-gcp-project-id.iam.gserviceaccount.com')

    if project == 'your-gcp-project-id':
        print("Please set GCP_PROJECT_ID, GCP_LOCATION_ID, GCP_TASK_QUEUE_ID, TASK_TARGET_URL, and TASK_SERVICE_ACCOUNT_EMAIL environment variables.")
    else:
        task_payload = {"message": "Hello from Cloud Tasks!", "source": "python-client"}
        try:
            created_task = create_http_task_with_auth(
                project,
                location,
                queue,
                target_url,
                sa_email,
                payload=task_payload,
                schedule_time_seconds=60 # Schedule 1 minute in the future
            )
            print(f"Successfully scheduled task: {created_task.name}")
        except Exception as e:
            print(f"Error creating task: {e}")
            print("Ensure Cloud Tasks API is enabled, queue exists, and authentication is set up.")