tessl/pypi-google-cloud-service-usage

Google Cloud Service Usage API client library for managing Google Cloud services programmatically

—

Pending

Overview

Eval results

Files

Operations Management

Name: tessl/pypi-google-cloud-service-usage
Author: tessl

Operations for managing and monitoring long-running service management tasks. Many service operations (enable, disable, batch operations) are asynchronous and return Operation objects that can be monitored for completion.

Capabilities

List Operations

Lists long-running operations associated with the Service Usage API, allowing you to monitor and track operation status.

def list_operations(
    request: ListOperationsRequest,
    *,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Optional[float] = None,
    metadata: Sequence[Tuple[str, str]] = ()
) -> ListOperationsPager:
    """
    List long-running operations.
    
    Args:
        request: The request object for listing operations
        retry: Retry configuration for the request
        timeout: Request timeout in seconds
        metadata: Additional metadata for the request
        
    Returns:
        Paginated iterator over Operation objects
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

Usage example:

from google.cloud import service_usage
from google.longrunning import operations_pb2

client = service_usage.ServiceUsageClient()

# List all operations
request = operations_pb2.ListOperationsRequest(
    name="projects/my-project-id/operations"
)

operations = client.list_operations(request=request)
for operation in operations:
    print(f"Operation: {operation.name}")
    print(f"Done: {operation.done}")
    if operation.done:
        print(f"Result: {operation.response}")
    else:
        print(f"Metadata: {operation.metadata}")

# List operations with filtering
filtered_request = operations_pb2.ListOperationsRequest(
    name="projects/my-project-id/operations",
    filter="done=false"  # Only show running operations
)

running_ops = client.list_operations(request=filtered_request)
print("Running operations:")
for op in running_ops:
    print(f"  {op.name}")

Get Operation

Retrieves detailed information about a specific long-running operation, including its current status and results.

def get_operation(
    request: GetOperationRequest,
    *,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Optional[float] = None,
    metadata: Sequence[Tuple[str, str]] = ()
) -> Operation:
    """
    Get details of a long-running operation.
    
    Args:
        request: The request object containing operation name
        retry: Retry configuration for the request
        timeout: Request timeout in seconds
        metadata: Additional metadata for the request
        
    Returns:
        Operation object with current status and result
        
    Raises:
        google.api_core.exceptions.NotFound: If the operation doesn't exist
        google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

Usage example:

from google.cloud import service_usage
from google.longrunning import operations_pb2

client = service_usage.ServiceUsageClient()

# Get a specific operation by name
operation_name = "projects/my-project-id/operations/operation-123456789"
request = operations_pb2.GetOperationRequest(name=operation_name)

operation = client.get_operation(request=request)
print(f"Operation: {operation.name}")
print(f"Done: {operation.done}")

if operation.done:
    if operation.HasField('response'):
        print("Operation completed successfully")
        print(f"Response: {operation.response}")
    elif operation.HasField('error'):
        print("Operation failed")
        print(f"Error: {operation.error.message}")
else:
    print("Operation still running")
    print(f"Metadata: {operation.metadata}")

Working with Long-Running Operations

Operation Monitoring

import time
from google.cloud import service_usage
from google.longrunning import operations_pb2

def monitor_operation(client, operation_name, timeout=300):
    """Monitor an operation until completion or timeout."""
    start_time = time.time()
    
    while time.time() - start_time < timeout:
        request = operations_pb2.GetOperationRequest(name=operation_name)
        operation = client.get_operation(request=request)
        
        if operation.done:
            if operation.HasField('response'):
                print("Operation completed successfully")
                return operation.response
            elif operation.HasField('error'):
                print(f"Operation failed: {operation.error.message}")
                return None
        
        print("Operation still running, waiting...")
        time.sleep(10)  # Wait 10 seconds before checking again
    
    print("Operation monitoring timed out")
    return None

# Usage
client = service_usage.ServiceUsageClient()

# Start a service enable operation
service_name = "projects/my-project-id/services/storage.googleapis.com"
enable_request = service_usage.EnableServiceRequest(name=service_name)
operation = client.enable_service(request=enable_request)

# Monitor the operation
result = monitor_operation(client, operation.name)
if result:
    print(f"Service enabled: {result}")

Using Operation.result()

The recommended way to wait for operation completion is using the result() method:

from google.cloud import service_usage

client = service_usage.ServiceUsageClient()

# Enable a service and wait for completion
service_name = "projects/my-project-id/services/storage.googleapis.com"
request = service_usage.EnableServiceRequest(name=service_name)

operation = client.enable_service(request=request)
print(f"Started operation: {operation.name}")

try:
    # Wait up to 5 minutes for completion
    result = operation.result(timeout=300)
    print(f"Service enabled successfully: {result.service.name}")
    print(f"Service state: {result.service.state}")
except Exception as e:
    print(f"Operation failed or timed out: {e}")

Handling Operation Errors

from google.cloud import service_usage
from google.api_core import exceptions

client = service_usage.ServiceUsageClient()

try:
    # Attempt to enable a service
    service_name = "projects/my-project-id/services/invalid-service.googleapis.com"
    request = service_usage.EnableServiceRequest(name=service_name)
    
    operation = client.enable_service(request=request)
    result = operation.result(timeout=60)
    
except exceptions.NotFound:
    print("Service not found")
except exceptions.PermissionDenied:
    print("Permission denied - check project permissions")
except exceptions.DeadlineExceeded:
    print("Operation timed out")
except Exception as e:
    print(f"Unexpected error: {e}")

Request Types

class ListOperationsRequest:
    """Request to list operations."""
    name: str  # Format: "projects/{project}/operations"
    filter: str  # Filter expression (e.g., "done=false")
    page_size: int  # Maximum number of operations per page
    page_token: str  # Token for pagination

class GetOperationRequest:
    """Request to get a specific operation."""
    name: str  # Full operation resource name

Response Types

from google.longrunning import operations_pb2
from google.rpc import status_pb2

class Operation:
    """A long-running operation."""
    name: str  # Operation resource name
    metadata: OperationMetadata  # Operation metadata
    done: bool  # Whether the operation is complete
    
    # One of the following will be set when done=True:
    error: status_pb2.Status  # Error information if operation failed
    response: Any  # Result if operation succeeded

class ListOperationsResponse:
    """Response from listing operations."""
    operations: List[Operation]  # List of operations
    next_page_token: str  # Token for next page

# Standard operations types
ListOperationsRequest = operations_pb2.ListOperationsRequest
GetOperationRequest = operations_pb2.GetOperationRequest
ListOperationsResponse = operations_pb2.ListOperationsResponse
Operation = operations_pb2.Operation
Status = status_pb2.Status

Operation Metadata

class OperationMetadata:
    """Metadata for Service Usage operations."""
    resource_names: List[str]  # Names of resources being operated on

The metadata provides information about which resources are being affected by the operation, helping you understand what services or configurations are being modified.

Best Practices

Always check operation completion: Use operation.result() or poll with get_operation()
Set appropriate timeouts: Service operations can take several minutes
Handle errors gracefully: Check for both request errors and operation failures
Monitor long-running operations: Use list_operations() to track multiple operations
Use batch operations: More efficient than individual operations when possible

Install with Tessl CLI

npx tessl i tessl/pypi-google-cloud-service-usage

docs

async-operations.md

index.md

operations-management.md

service-discovery.md

service-management.md

tile.json

tessl/pypi-google-cloud-service-usage