tessl/pypi-google-cloud-tasks
Google Cloud Tasks API client library for managing distributed task queues.
—
Pending
client-operations.mddocs/
Client Operations
Core client functionality for managing Cloud Tasks service connections, authentication, and transport configuration. Provides both synchronous and asynchronous clients with identical APIs.
Capabilities
CloudTasksClient
Synchronous client for Cloud Tasks service operations with automatic authentication and transport management.
class CloudTasksClient:
def __init__(
self,
*,
credentials: ga_credentials.Credentials = None,
transport: Union[str, CloudTasksTransport] = None,
client_options: client_options_lib.ClientOptions = None,
client_info: gapic_v1.client_info.ClientInfo = None
):
"""Initialize the CloudTasksClient.
Args:
credentials: The authorization credentials to attach to requests.
transport: The transport to use for API calls.
client_options: Custom options for the client.
client_info: The client info used to send a user-agent string.
"""
@classmethod
def from_service_account_file(
cls,
filename: str,
*args,
**kwargs
) -> "CloudTasksClient":
"""Create a client from a service account JSON file.
Args:
filename: Path to the service account JSON file.
Returns:
The constructed client.
"""
@classmethod
def from_service_account_info(
cls,
info: dict,
*args,
**kwargs
) -> "CloudTasksClient":
"""Create a client from service account info.
Args:
info: The service account info in Google format.
Returns:
The constructed client.
"""
@classmethod
def get_transport_class(
cls,
label: Optional[str] = None
) -> Type[CloudTasksTransport]:
"""Return an appropriate transport class.
Args:
label: The name of the desired transport.
Returns:
The transport class to use.
"""
@property
def transport(self) -> CloudTasksTransport:
"""Return the transport used by the client."""
@property
def api_endpoint(self) -> str:
"""Return the API endpoint used by the client."""
@property
def universe_domain(self) -> str:
"""Return the universe domain used by the client."""
def __enter__(self) -> "CloudTasksClient":
"""Enter the context manager."""
def __exit__(self, type, value, traceback) -> None:
"""Exit the context manager."""CloudTasksAsyncClient
Asynchronous client for Cloud Tasks service operations with async/await support and identical API to the synchronous client.
class CloudTasksAsyncClient:
def __init__(
self,
*,
credentials: ga_credentials.Credentials = None,
transport: Union[str, CloudTasksAsyncTransport] = None,
client_options: client_options_lib.ClientOptions = None,
client_info: gapic_v1.client_info.ClientInfo = None
):
"""Initialize the CloudTasksAsyncClient.
Args:
credentials: The authorization credentials to attach to requests.
transport: The transport to use for API calls.
client_options: Custom options for the client.
client_info: The client info used to send a user-agent string.
"""
@classmethod
def from_service_account_file(
cls,
filename: str,
*args,
**kwargs
) -> "CloudTasksAsyncClient":
"""Create an async client from a service account JSON file."""
@classmethod
def from_service_account_info(
cls,
info: dict,
*args,
**kwargs
) -> "CloudTasksAsyncClient":
"""Create an async client from service account info."""
async def __aenter__(self) -> "CloudTasksAsyncClient":
"""Enter the async context manager."""
async def __aexit__(self, exc_type, exc, tb) -> None:
"""Exit the async context manager."""Path Utilities
Static methods for constructing and parsing Google Cloud resource paths.
@staticmethod
def queue_path(project: str, location: str, queue: str) -> str:
"""Return a fully-qualified queue string.
Args:
project: The project ID.
location: The location ID.
queue: The queue ID.
Returns:
The queue path string.
"""
@staticmethod
def parse_queue_path(path: str) -> Dict[str, str]:
"""Parse a queue path into its component segments.
Args:
path: A fully-qualified queue path.
Returns:
A dictionary with project, location, and queue keys.
"""
@staticmethod
def task_path(project: str, location: str, queue: str, task: str) -> str:
"""Return a fully-qualified task string.
Args:
project: The project ID.
location: The location ID.
queue: The queue ID.
task: The task ID.
Returns:
The task path string.
"""
@staticmethod
def parse_task_path(path: str) -> Dict[str, str]:
"""Parse a task path into its component segments.
Args:
path: A fully-qualified task path.
Returns:
A dictionary with project, location, queue, and task keys.
"""
@staticmethod
def common_project_path(project: str) -> str:
"""Return a fully-qualified project string."""
@staticmethod
def parse_common_project_path(path: str) -> Dict[str, str]:
"""Parse a project path into its component segments."""
@staticmethod
def common_location_path(project: str, location: str) -> str:
"""Return a fully-qualified location string."""
@staticmethod
def parse_common_location_path(path: str) -> Dict[str, str]:
"""Parse a location path into its component segments."""Usage Examples
Basic Client Creation
from google.cloud import tasks
# Create a synchronous client with default credentials
client = tasks.CloudTasksClient()
# Create from service account file
client = tasks.CloudTasksClient.from_service_account_file('path/to/service-account.json')
# Create from service account info dictionary
service_account_info = {...} # Service account JSON data
client = tasks.CloudTasksClient.from_service_account_info(service_account_info)Asynchronous Client Usage
import asyncio
from google.cloud import tasks
async def main():
# Create an async client
async with tasks.CloudTasksAsyncClient() as client:
# Use client for async operations
queues = await client.list_queues(parent='projects/my-project/locations/us-central1')
async for queue in queues:
print(f'Queue: {queue.name}')
asyncio.run(main())Path Construction
from google.cloud import tasks
client = tasks.CloudTasksClient()
# Construct resource paths
project = 'my-project-id'
location = 'us-central1'
queue_name = 'my-queue'
task_name = 'my-task'
# Build paths
parent_path = client.common_location_path(project, location)
queue_path = client.queue_path(project, location, queue_name)
task_path = client.task_path(project, location, queue_name, task_name)
print(f'Parent: {parent_path}')
print(f'Queue: {queue_path}')
print(f'Task: {task_path}')
# Parse paths
queue_parts = client.parse_queue_path(queue_path)
print(f'Project: {queue_parts["project"]}')
print(f'Location: {queue_parts["location"]}')
print(f'Queue: {queue_parts["queue"]}')Install with Tessl CLI
npx tessl i tessl/pypi-google-cloud-tasks