Google Cloud Billing Budgets API client library for creating and managing cloud billing budgets
—
Client classes for interacting with the Google Cloud Billing Budgets API. Provides both synchronous and asynchronous implementations with comprehensive support for all budget operations.
The main synchronous client for budget operations with full CRUD support and built-in authentication.
class BudgetServiceClient:
def __init__(
self,
credentials: Optional[Credentials] = None,
transport: Optional[Union[str, BudgetServiceTransport]] = None,
client_options: Optional[ClientOptions] = None,
client_info: Optional[gapic_v1.client_info.ClientInfo] = None,
**kwargs
):
"""
Initialize the Budget Service Client.
Args:
credentials: The authorization credentials to attach to requests
transport: The transport to use ('grpc', 'grpc_asyncio', 'rest')
client_options: Custom options for the client
client_info: The client info used to send user agent headers
"""
def create_budget(
self,
request: Optional[Union[CreateBudgetRequest, dict]] = None,
*,
parent: Optional[str] = None,
budget: Optional[Budget] = None,
retry: OptionalRetry = None,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> Budget:
"""
Creates a budget and returns the created budget.
Args:
request: The request object or dict
parent: Required. Name of the billing account (billingAccounts/{id})
budget: Required. Budget to create
retry: Retry configuration
timeout: Timeout in seconds
metadata: Additional gRPC metadata
Returns:
Budget: The created budget
Raises:
google.api_core.exceptions.GoogleAPICallError: If the request fails
"""
def get_budget(
self,
request: Optional[Union[GetBudgetRequest, dict]] = None,
*,
name: Optional[str] = None,
retry: OptionalRetry = None,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> Budget:
"""
Returns a budget.
Args:
request: The request object or dict
name: Required. Budget name (billingAccounts/{id}/budgets/{id})
retry: Retry configuration
timeout: Timeout in seconds
metadata: Additional gRPC metadata
Returns:
Budget: The requested budget
Raises:
google.api_core.exceptions.NotFound: If budget doesn't exist
"""
def list_budgets(
self,
request: Optional[Union[ListBudgetsRequest, dict]] = None,
*,
parent: Optional[str] = None,
scope: Optional[str] = None,
retry: OptionalRetry = None,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> ListBudgetsPager:
"""
Returns a list of budgets for a billing account.
Args:
request: The request object or dict
parent: Required. Name of the billing account (billingAccounts/{id})
scope: Optional. Set to projects/{id} to list budgets for project
retry: Retry configuration
timeout: Timeout in seconds
metadata: Additional gRPC metadata
Returns:
ListBudgetsPager: Pager for iterating through results
Raises:
google.api_core.exceptions.GoogleAPICallError: If the request fails
"""
def update_budget(
self,
request: Optional[Union[UpdateBudgetRequest, dict]] = None,
*,
budget: Optional[Budget] = None,
update_mask: Optional[FieldMask] = None,
retry: OptionalRetry = None,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> Budget:
"""
Updates a budget and returns the updated budget.
Args:
request: The request object or dict
budget: Required. Updated budget
update_mask: Optional. Indicates which fields to update
retry: Retry configuration
timeout: Timeout in seconds
metadata: Additional gRPC metadata
Returns:
Budget: The updated budget
Raises:
google.api_core.exceptions.NotFound: If budget doesn't exist
"""
def delete_budget(
self,
request: Optional[Union[DeleteBudgetRequest, dict]] = None,
*,
name: Optional[str] = None,
retry: OptionalRetry = None,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> None:
"""
Deletes a budget.
Args:
request: The request object or dict
name: Required. Budget name (billingAccounts/{id}/budgets/{id})
retry: Retry configuration
timeout: Timeout in seconds
metadata: Additional gRPC metadata
Raises:
google.api_core.exceptions.NotFound: If budget doesn't exist
"""Fully asynchronous client with the same API surface as the synchronous client but using async/await patterns.
class BudgetServiceAsyncClient:
def __init__(
self,
credentials: Optional[Credentials] = None,
transport: Optional[Union[str, BudgetServiceTransport]] = None,
client_options: Optional[ClientOptions] = None,
client_info: Optional[gapic_v1.client_info.ClientInfo] = None,
**kwargs
):
"""Initialize the async Budget Service Client."""
async def create_budget(
self,
request: Optional[Union[CreateBudgetRequest, dict]] = None,
*,
parent: Optional[str] = None,
budget: Optional[Budget] = None,
retry: OptionalRetry = None,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> Budget:
"""Async version of create_budget."""
async def get_budget(
self,
request: Optional[Union[GetBudgetRequest, dict]] = None,
*,
name: Optional[str] = None,
retry: OptionalRetry = None,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> Budget:
"""Async version of get_budget."""
async def list_budgets(
self,
request: Optional[Union[ListBudgetsRequest, dict]] = None,
*,
parent: Optional[str] = None,
scope: Optional[str] = None,
retry: OptionalRetry = None,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> ListBudgetsAsyncPager:
"""Async version of list_budgets."""
async def update_budget(
self,
request: Optional[Union[UpdateBudgetRequest, dict]] = None,
*,
budget: Optional[Budget] = None,
update_mask: Optional[FieldMask] = None,
retry: OptionalRetry = None,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> Budget:
"""Async version of update_budget."""
async def delete_budget(
self,
request: Optional[Union[DeleteBudgetRequest, dict]] = None,
*,
name: Optional[str] = None,
retry: OptionalRetry = None,
timeout: Optional[float] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> None:
"""Async version of delete_budget."""Both client classes provide factory methods for authentication and path construction utilities.
# Available on both BudgetServiceClient and BudgetServiceAsyncClient
@classmethod
def from_service_account_info(
cls,
info: dict,
*args,
**kwargs
) -> BudgetServiceClient:
"""
Create client from service account info dictionary.
Args:
info: Service account info in Google format
Returns:
BudgetServiceClient: Configured client instance
"""
@classmethod
def from_service_account_file(
cls,
filename: str,
*args,
**kwargs
) -> BudgetServiceClient:
"""
Create client from service account JSON file.
Args:
filename: Path to service account JSON file
Returns:
BudgetServiceClient: Configured client instance
"""Utility methods for constructing and parsing Google Cloud resource names.
# Path construction methods
@staticmethod
def budget_path(billing_account: str, budget: str) -> str:
"""Return a fully-qualified budget string."""
@staticmethod
def parse_budget_path(path: str) -> Dict[str, str]:
"""Parse a budget path into its component segments."""
@staticmethod
def common_billing_account_path(billing_account: str) -> str:
"""Return a fully-qualified billing_account string."""
@staticmethod
def parse_common_billing_account_path(path: str) -> Dict[str, str]:
"""Parse a billing_account path into its component segments."""
@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_organization_path(organization: str) -> str:
"""Return a fully-qualified organization string."""
@staticmethod
def parse_common_organization_path(path: str) -> Dict[str, str]:
"""Parse an organization path into its component segments."""
@staticmethod
def common_folder_path(folder: str) -> str:
"""Return a fully-qualified folder string."""
@staticmethod
def parse_common_folder_path(path: str) -> Dict[str, str]:
"""Parse a folder 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."""Specialized pager classes for handling paginated responses from list operations.
class ListBudgetsPager:
"""Synchronous pager for list_budgets results."""
def __iter__(self) -> Iterator[Budget]:
"""Iterate over Budget objects."""
@property
def pages(self) -> Iterator[ListBudgetsResponse]:
"""Iterate over raw response pages."""
class ListBudgetsAsyncPager:
"""Asynchronous pager for list_budgets results."""
def __aiter__(self) -> AsyncIterator[Budget]:
"""Async iterate over Budget objects."""
@property
def pages(self) -> AsyncIterator[ListBudgetsResponse]:
"""Async iterate over raw response pages."""from google.cloud.billing import budgets
# Initialize client with default credentials
client = budgets.BudgetServiceClient()
# Or with custom credentials
from google.oauth2 import service_account
credentials = service_account.Credentials.from_service_account_file(
"path/to/service-account.json"
)
client = budgets.BudgetServiceClient(credentials=credentials)
# Or from environment
client = budgets.BudgetServiceClient.from_service_account_file(
"path/to/service-account.json"
)import asyncio
from google.cloud.billing import budgets
async def manage_budgets():
client = budgets.BudgetServiceAsyncClient()
parent = "billingAccounts/123456-ABCDEF-789012"
# List budgets asynchronously
async for budget in await client.list_budgets(parent=parent):
print(f"Budget: {budget.display_name}")
await client.close()
# Run the async function
asyncio.run(manage_budgets())from google.cloud.billing import budgets
client = budgets.BudgetServiceClient()
parent = "billingAccounts/123456-ABCDEF-789012"
# Iterate through all budgets
for budget in client.list_budgets(parent=parent):
print(f"Budget: {budget.display_name}")
# Or iterate through pages
pager = client.list_budgets(parent=parent)
for page in pager.pages:
print(f"Page has {len(page.budgets)} budgets")
for budget in page.budgets:
print(f" - {budget.display_name}")Install with Tessl CLI
npx tessl i tessl/pypi-google-cloud-billing-budgets