CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-google-cloud-tasks

Google Cloud Tasks API client library for managing distributed task queues.

—

Pending
Overview
Eval results
Files

task-targeting.mddocs/

Task Targeting

Task target configuration for HTTP endpoints and App Engine applications, including authentication, routing, and request formatting options.

Capabilities

HTTP Request Targeting

Configure tasks to make HTTP requests to external endpoints with full control over method, headers, body, and authentication.

class HttpRequest:
    url: str  # Required full URL path for request
    http_method: HttpMethod  # HTTP method to use (default POST)
    headers: MutableMapping[str, str]  # HTTP request headers
    body: bytes  # HTTP request body (allowed only for POST/PUT/PATCH)
    oauth_token: OAuthToken  # OAuth token for authorization (oneof authorization_header)
    oidc_token: OidcToken  # OIDC token for authorization (oneof authorization_header)

App Engine Request Targeting

Configure tasks to call App Engine applications with service/version routing and relative URI paths.

class AppEngineHttpRequest:
    http_method: HttpMethod  # HTTP method to use (default POST)
    app_engine_routing: AppEngineRouting  # Task-level App Engine routing settings
    relative_uri: str  # Relative URI (must begin with "/")
    headers: MutableMapping[str, str]  # HTTP request headers
    body: bytes  # HTTP request body (allowed only for POST/PUT)

App Engine Routing

Control which App Engine service, version, and instance processes the task.

class AppEngineRouting:
    service: str  # App service (default service if not specified)
    version: str  # App version (default version if not specified)
    instance: str  # App instance (available instance if not specified)
    host: str  # Output only host that task is sent to

OAuth Authentication

Configure OAuth tokens for authenticating requests to Google Cloud APIs and services.

class OAuthToken:
    service_account_email: str  # Service account email for generating OAuth token
    scope: str  # OAuth scope (default "https://www.googleapis.com/auth/cloud-platform")

OIDC Authentication

Configure OpenID Connect tokens for authenticating requests to any HTTP endpoint.

class OidcToken:
    service_account_email: str  # Service account email for generating OIDC token
    audience: str  # Audience for OIDC token (uses target URI if not specified)

HTTP Methods

Supported HTTP methods for task requests.

class HttpMethod:
    HTTP_METHOD_UNSPECIFIED = 0  # HTTP method unspecified
    POST = 1  # HTTP POST
    GET = 2  # HTTP GET
    HEAD = 3  # HTTP HEAD
    PUT = 4  # HTTP PUT
    DELETE = 5  # HTTP DELETE
    PATCH = 6  # HTTP PATCH
    OPTIONS = 7  # HTTP OPTIONS

Usage Examples

Basic HTTP Tasks

from google.cloud import tasks
import json

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'http-queue')

# Simple POST request
simple_task = tasks.Task(
    http_request=tasks.HttpRequest(
        url='https://api.example.com/webhook',
        http_method=tasks.HttpMethod.POST,
        body=b'{"event": "task_executed"}'
    )
)

# GET request with query parameters (include in URL)
get_task = tasks.Task(
    http_request=tasks.HttpRequest(
        url='https://api.example.com/status?check=health',
        http_method=tasks.HttpMethod.GET
    )
)

# Create both tasks
client.create_task(parent=queue_path, task=simple_task)
client.create_task(parent=queue_path, task=get_task)

HTTP Tasks with Headers and JSON Payload

from google.cloud import tasks
import json

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'api-queue')

# Structured JSON payload with custom headers
payload = {
    'user_id': 12345,
    'action': 'process_order',
    'metadata': {
        'priority': 'high',
        'source': 'cloud_tasks'
    }
}

task = tasks.Task(
    http_request=tasks.HttpRequest(
        url='https://myapp.com/api/process',
        http_method=tasks.HttpMethod.POST,
        headers={
            'Content-Type': 'application/json',
            'X-Custom-Header': 'task-processing',
            'Authorization': 'Bearer api-key-123'  # Static auth (not recommended for production)
        },
        body=json.dumps(payload).encode('utf-8')
    )
)

created_task = client.create_task(parent=queue_path, task=task)
print(f'Created API task: {created_task.name}')

Authenticated HTTP Tasks with OAuth

from google.cloud import tasks

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'secure-queue')

# OAuth-authenticated request to Google Cloud API
oauth_task = tasks.Task(
    http_request=tasks.HttpRequest(
        url='https://cloudfunctions.googleapis.com/v1/projects/my-project/locations/us-central1/functions/processor:call',
        http_method=tasks.HttpMethod.POST,
        oauth_token=tasks.OAuthToken(
            service_account_email='task-runner@my-project.iam.gserviceaccount.com',
            scope='https://www.googleapis.com/auth/cloud-platform'
        ),
        headers={'Content-Type': 'application/json'},
        body=b'{"data": "secure_operation"}'
    )
)

created_task = client.create_task(parent=queue_path, task=oauth_task)
print(f'Created OAuth task: {created_task.name}')

Authenticated HTTP Tasks with OIDC

from google.cloud import tasks

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'oidc-queue')

# OIDC-authenticated request to custom endpoint
oidc_task = tasks.Task(
    http_request=tasks.HttpRequest(
        url='https://secure-api.mycompany.com/process',
        http_method=tasks.HttpMethod.POST,
        oidc_token=tasks.OidcToken(
            service_account_email='task-runner@my-project.iam.gserviceaccount.com',
            audience='https://secure-api.mycompany.com'
        ),
        headers={'Content-Type': 'application/json'},
        body=b'{"secure": "data"}'
    )
)

created_task = client.create_task(parent=queue_path, task=oidc_task)
print(f'Created OIDC task: {created_task.name}')

App Engine Tasks

from google.cloud import tasks

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'appengine-queue')

# Basic App Engine task
basic_ae_task = tasks.Task(
    app_engine_http_request=tasks.AppEngineHttpRequest(
        http_method=tasks.HttpMethod.POST,
        relative_uri='/tasks/process',
        body=b'{"task": "basic_processing"}'
    )
)

# App Engine task with specific routing
routed_ae_task = tasks.Task(
    app_engine_http_request=tasks.AppEngineHttpRequest(
        http_method=tasks.HttpMethod.POST,
        relative_uri='/api/heavy-computation',
        app_engine_routing=tasks.AppEngineRouting(
            service='worker-service',
            version='v2'
        ),
        headers={'Content-Type': 'application/json'},
        body=b'{"computation": "complex_algorithm"}'
    )
)

# Create both tasks
client.create_task(parent=queue_path, task=basic_ae_task)
client.create_task(parent=queue_path, task=routed_ae_task)

App Engine Tasks with Custom Headers

from google.cloud import tasks
import json

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'ae-custom-queue')

# App Engine task with custom headers and routing
task_data = {
    'batch_id': 'batch_001',
    'items': ['item1', 'item2', 'item3']
}

custom_ae_task = tasks.Task(
    app_engine_http_request=tasks.AppEngineHttpRequest(
        http_method=tasks.HttpMethod.POST,
        relative_uri='/batch/process',
        app_engine_routing=tasks.AppEngineRouting(
            service='batch-processor',
            version='stable',
            instance='instance-1'  # Target specific instance
        ),
        headers={
            'Content-Type': 'application/json',
            'X-Batch-Priority': 'high',
            'X-Processing-Mode': 'parallel'
        },
        body=json.dumps(task_data).encode('utf-8')
    )
)

created_task = client.create_task(parent=queue_path, task=custom_ae_task)
print(f'Created custom App Engine task: {created_task.name}')

Multi-Method HTTP Task Examples

from google.cloud import tasks

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'multi-method-queue')

# PUT request for resource updates
put_task = tasks.Task(
    http_request=tasks.HttpRequest(
        url='https://api.example.com/resources/123',
        http_method=tasks.HttpMethod.PUT,
        headers={'Content-Type': 'application/json'},
        body=b'{"status": "updated", "timestamp": "2023-01-01T00:00:00Z"}'
    )
)

# DELETE request for cleanup
delete_task = tasks.Task(
    http_request=tasks.HttpRequest(
        url='https://api.example.com/temp-resources/temp-123',
        http_method=tasks.HttpMethod.DELETE,
        headers={'Authorization': 'Bearer cleanup-token'}
    )
)

# PATCH request for partial updates
patch_task = tasks.Task(
    http_request=tasks.HttpRequest(
        url='https://api.example.com/users/456',
        http_method=tasks.HttpMethod.PATCH,
        headers={'Content-Type': 'application/json'},
        body=b'{"last_login": "2023-01-01T12:00:00Z"}'
    )
)

# Create all tasks
client.create_task(parent=queue_path, task=put_task)
client.create_task(parent=queue_path, task=delete_task)
client.create_task(parent=queue_path, task=patch_task)

Error Handling and Retry Considerations

from google.cloud import tasks
from google.protobuf import duration_pb2

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'resilient-queue')

# Task with custom dispatch deadline
resilient_task = tasks.Task(
    http_request=tasks.HttpRequest(
        url='https://slow-api.example.com/process',
        http_method=tasks.HttpMethod.POST,
        body=b'{"data": "long_running_operation"}'
    ),
    dispatch_deadline=duration_pb2.Duration(seconds=300)  # 5 minute timeout
)

created_task = client.create_task(parent=queue_path, task=resilient_task)
print(f'Created resilient task with 5min timeout: {created_task.name}')

# The queue's retry configuration will handle retries automatically
# if the endpoint returns 5xx errors or times out

Install with Tessl CLI

npx tessl i tessl/pypi-google-cloud-tasks

docs

client-operations.md

iam-security.md

index.md

queue-configuration.md

queue-management.md

task-management.md

task-targeting.md

tile.json