tessl/pypi-google-cloud-service-usage
Google Cloud Service Usage API client library for managing Google Cloud services programmatically
—
Pending
async-operations.mddocs/
Asynchronous Client Operations
Asynchronous versions of all service management and discovery operations for non-blocking I/O in async applications. The async client provides the same functionality as the synchronous client but with async/await support.
Client Initialization
class ServiceUsageAsyncClient:
"""Async client for Google Cloud Service Usage API."""
def __init__(
self,
*,
credentials: Optional[ga_credentials.Credentials] = None,
transport: Optional[Union[str, ServiceUsageTransport]] = None,
client_options: Optional[ClientOptions] = None,
client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO
):
"""Initialize the async client."""
async def __aenter__(self) -> "ServiceUsageAsyncClient":
"""Async context manager entry."""
async def __aexit__(self, exc_type, exc, tb):
"""Async context manager exit."""Usage example:
import asyncio
from google.cloud import service_usage
async def main():
async with service_usage.ServiceUsageAsyncClient() as client:
# Use the client for async operations
pass
# Run the async function
asyncio.run(main())Capabilities
Async Service Management
Asynchronous versions of service enable, disable, and batch operations.
async def enable_service(
request: EnableServiceRequest,
*,
retry: OptionalAsyncRetry = gapic_v1.method.DEFAULT,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> AsyncOperation[EnableServiceResponse, OperationMetadata]:
"""
Asynchronously enable a service for a project.
Args:
request: The request object containing the service name
retry: Async retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata for the request
Returns:
Async long-running operation that resolves to EnableServiceResponse
"""
async def disable_service(
request: DisableServiceRequest,
*,
retry: OptionalAsyncRetry = gapic_v1.method.DEFAULT,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> AsyncOperation[DisableServiceResponse, OperationMetadata]:
"""
Asynchronously disable a service for a project.
Args:
request: The request object containing the service name and options
retry: Async retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata for the request
Returns:
Async long-running operation that resolves to DisableServiceResponse
"""
async def batch_enable_services(
request: BatchEnableServicesRequest,
*,
retry: OptionalAsyncRetry = gapic_v1.method.DEFAULT,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> AsyncOperation[BatchEnableServicesResponse, OperationMetadata]:
"""
Asynchronously enable multiple services in a single operation.
Args:
request: The request object containing parent and service IDs
retry: Async retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata for the request
Returns:
Async long-running operation that resolves to BatchEnableServicesResponse
"""Usage example:
import asyncio
from google.cloud import service_usage
async def enable_services_async():
async with service_usage.ServiceUsageAsyncClient() as client:
# Enable a single service
service_name = "projects/my-project-id/services/storage.googleapis.com"
request = service_usage.EnableServiceRequest(name=service_name)
operation = await client.enable_service(request=request)
result = await operation.result()
print(f"Service enabled: {result.service.name}")
# Batch enable multiple services
parent = "projects/my-project-id"
batch_request = service_usage.BatchEnableServicesRequest(
parent=parent,
service_ids=["compute.googleapis.com", "bigquery.googleapis.com"]
)
batch_operation = await client.batch_enable_services(request=batch_request)
batch_result = await batch_operation.result()
print(f"Enabled {len(batch_result.services)} services")
asyncio.run(enable_services_async())Async Service Discovery
Asynchronous versions of service query and discovery operations.
async def get_service(
request: GetServiceRequest,
*,
retry: OptionalAsyncRetry = gapic_v1.method.DEFAULT,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> Service:
"""
Asynchronously get information about a specific service.
Args:
request: The request object containing the service name
retry: Async retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata for the request
Returns:
Service object with configuration details
"""
async def list_services(
request: ListServicesRequest,
*,
retry: OptionalAsyncRetry = gapic_v1.method.DEFAULT,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> ListServicesAsyncPager:
"""
Asynchronously list services available to a project.
Args:
request: The request object with parent and filtering options
retry: Async retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata for the request
Returns:
Async paginated iterator over Service objects
"""
async def batch_get_services(
request: BatchGetServicesRequest,
*,
retry: OptionalAsyncRetry = gapic_v1.method.DEFAULT,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> BatchGetServicesResponse:
"""
Asynchronously get information about multiple services.
Args:
request: The request object containing parent and service names
retry: Async retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata for the request
Returns:
BatchGetServicesResponse with list of services
"""Usage example:
import asyncio
from google.cloud import service_usage
async def discover_services_async():
async with service_usage.ServiceUsageAsyncClient() as client:
# Get a specific service
service_name = "projects/my-project-id/services/storage.googleapis.com"
request = service_usage.GetServiceRequest(name=service_name)
service = await client.get_service(request=request)
print(f"Service: {service.config.title}")
print(f"State: {service.state.name}")
# List all services asynchronously
parent = "projects/my-project-id"
list_request = service_usage.ListServicesRequest(parent=parent)
async for service in await client.list_services(request=list_request):
print(f"Service: {service.name} - {service.state.name}")
# Batch get multiple services
service_names = [
f"{parent}/services/storage.googleapis.com",
f"{parent}/services/compute.googleapis.com"
]
batch_request = service_usage.BatchGetServicesRequest(
parent=parent,
names=service_names
)
batch_response = await client.batch_get_services(request=batch_request)
for service in batch_response.services:
print(f"Batch result: {service.config.title}")
asyncio.run(discover_services_async())Async Operations Management
Asynchronous operations for managing long-running tasks.
async def list_operations(
request: ListOperationsRequest,
*,
retry: OptionalAsyncRetry = gapic_v1.method.DEFAULT,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> ListOperationsAsyncPager:
"""
Asynchronously list long-running operations.
Args:
request: The request object for listing operations
retry: Async retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata for the request
Returns:
Async paginated iterator over Operation objects
"""
async def get_operation(
request: GetOperationRequest,
*,
retry: OptionalAsyncRetry = gapic_v1.method.DEFAULT,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> Operation:
"""
Asynchronously get details of a long-running operation.
Args:
request: The request object containing operation name
retry: Async retry configuration
timeout: Request timeout in seconds
metadata: Additional metadata for the request
Returns:
Operation object with current status and result
"""Async Pager
class ListServicesAsyncPager:
"""Async pager for iterating through list_services results."""
def __aiter__(self) -> AsyncIterator[Service]:
"""Asynchronously iterate through Service objects."""
@property
def pages(self) -> AsyncIterator[ListServicesResponse]:
"""Asynchronously iterate through response pages."""Usage example:
import asyncio
from google.cloud import service_usage
async def paginate_services_async():
async with service_usage.ServiceUsageAsyncClient() as client:
parent = "projects/my-project-id"
request = service_usage.ListServicesRequest(
parent=parent,
page_size=5
)
pager = await client.list_services(request=request)
# Iterate through individual services
async for service in pager:
print(f"Service: {service.name}")
# Or iterate through pages
async for page in pager.pages:
print(f"Page with {len(page.services)} services")
for service in page.services:
print(f" {service.config.title}")
asyncio.run(paginate_services_async())Async Context Manager Pattern
import asyncio
from google.cloud import service_usage
async def async_context_example():
# Using async context manager for automatic cleanup
async with service_usage.ServiceUsageAsyncClient() as client:
# All async operations here
parent = "projects/my-project-id"
request = service_usage.ListServicesRequest(parent=parent)
services = []
async for service in await client.list_services(request=request):
services.append(service)
print(f"Found {len(services)} services")
# Client is automatically closed when exiting the context
asyncio.run(async_context_example())Install with Tessl CLI
npx tessl i tessl/pypi-google-cloud-service-usage