Common protocol buffer definitions used across Google APIs and client libraries
—
Cloud-specific extensions for location services, extended operations, and Google Cloud Platform integration. Provides types for multi-region services, cloud resource management, and platform-specific functionality.
Manage and query cloud service locations and regions.
from google.cloud.location.locations_pb2 import (
ListLocationsRequest, ListLocationsResponse,
GetLocationRequest, Location
)
class ListLocationsRequest(message.Message):
"""Request to list available service locations."""
name: str # Parent resource name (e.g., "projects/my-project")
filter: str # Filter expression for locations
page_size: int # Maximum number of locations to return
page_token: str # Page token for pagination
class ListLocationsResponse(message.Message):
"""Response containing available locations."""
locations: list[Location] # List of available locations
next_page_token: str # Token for next page (if any)
class GetLocationRequest(message.Message):
"""Request to get details of a specific location."""
name: str # Location resource name
class Location(message.Message):
"""Cloud service location information."""
name: str # Location resource name
location_id: str # Location identifier (e.g., "us-central1")
display_name: str # Human-readable location name
labels: dict[str, str] # Location labels and metadata
metadata: Any # Location-specific metadataExtended operation support for complex cloud operations.
from google.cloud.extended_operations_pb2 import ExtendedOperation
class ExtendedOperation(message.Message):
"""Extended operation with additional cloud-specific fields."""
# Inherits basic operation fields and adds:
# Additional cloud-specific operation metadata
# Enhanced status reporting
# Region and zone information
# Resource hierarchy contextMetadata for generating cloud service client libraries.
from google.gapic.metadata.gapic_metadata_pb2 import (
GapicMetadata, ServiceForTransport, ServiceAsClient, MethodList
)
class GapicMetadata(message.Message):
"""Metadata for GAPIC (Generated API Client) libraries."""
schema: str # Metadata schema version
comment: str # Additional comments
language: str # Target programming language
proto_package: str # Protocol buffer package name
library_package: str # Generated library package name
services: dict[str, ServiceForTransport] # Service transport mappings
class ServiceForTransport(message.Message):
"""Service configuration for different transports."""
clients: dict[str, ServiceAsClient] # Client configurations
class ServiceAsClient(message.Message):
"""Service as client configuration."""
library_client: str # Client class name
rpcs: dict[str, MethodList] # RPC method mappings
class MethodList(message.Message):
"""List of methods for a service."""
methods: list[str] # Method namesfrom google.cloud.location.locations_pb2 import ListLocationsRequest, ListLocationsResponse
def list_all_locations(parent: str, client) -> list:
"""List all available locations for a service."""
request = ListLocationsRequest()
request.name = parent # e.g., "projects/my-project"
request.page_size = 100
all_locations = []
while True:
response: ListLocationsResponse = client.list_locations(request)
all_locations.extend(response.locations)
# Check if there are more pages
if not response.next_page_token:
break
# Set up next page request
request.page_token = response.next_page_token
return all_locations
# Example usage
locations = list_all_locations("projects/my-project", location_client)
for location in locations:
print(f"Location: {location.location_id} ({location.display_name})")from google.cloud.location.locations_pb2 import ListLocationsRequest
def list_locations_by_region(parent: str, client, region_filter: str = None):
"""List locations with optional region filtering."""
request = ListLocationsRequest()
request.name = parent
if region_filter:
# Filter by region (e.g., "us-", "europe-", "asia-")
request.filter = f'labels.region:"{region_filter}*"'
response = client.list_locations(request)
return response.locations
# Examples
us_locations = list_locations_by_region("projects/my-project", client, "us")
europe_locations = list_locations_by_region("projects/my-project", client, "europe")from google.cloud.location.locations_pb2 import GetLocationRequest, Location
def get_location_info(location_name: str, client) -> Location:
"""Get detailed information about a specific location."""
request = GetLocationRequest()
request.name = location_name # e.g., "projects/my-project/locations/us-central1"
location = client.get_location(request)
return location
def print_location_details(location: Location):
"""Print detailed location information."""
print(f"Location Name: {location.name}")
print(f"Location ID: {location.location_id}")
print(f"Display Name: {location.display_name}")
print("Labels:")
for key, value in location.labels.items():
print(f" {key}: {value}")
if location.metadata:
print(f"Metadata: {location.metadata}")
# Example usage
location = get_location_info("projects/my-project/locations/us-central1", client)
print_location_details(location)from google.cloud.location.locations_pb2 import Location
def select_optimal_location(locations: list[Location], preferences: dict) -> Location:
"""Select optimal location based on preferences."""
scored_locations = []
for location in locations:
score = 0
# Score based on region preference
if preferences.get("preferred_region"):
if location.location_id.startswith(preferences["preferred_region"]):
score += 10
# Score based on labels
for label_key, preferred_value in preferences.get("labels", {}).items():
if location.labels.get(label_key) == preferred_value:
score += 5
# Score based on availability zones (if in metadata)
if preferences.get("multi_zone") and "zones" in location.labels:
zone_count = len(location.labels["zones"].split(","))
score += zone_count
scored_locations.append((location, score))
# Return location with highest score
if scored_locations:
return max(scored_locations, key=lambda x: x[1])[0]
return locations[0] if locations else None
# Example usage
preferences = {
"preferred_region": "us-central",
"labels": {
"tier": "premium"
},
"multi_zone": True
}
optimal_location = select_optimal_location(locations, preferences)
print(f"Selected location: {optimal_location.location_id}")from google.gapic.metadata.gapic_metadata_pb2 import GapicMetadata
def parse_gapic_metadata(metadata: GapicMetadata):
"""Parse and display GAPIC metadata information."""
print(f"Schema: {metadata.schema}")
print(f"Language: {metadata.language}")
print(f"Proto Package: {metadata.proto_package}")
print(f"Library Package: {metadata.library_package}")
print("\nServices:")
for service_name, service_transport in metadata.services.items():
print(f" Service: {service_name}")
for transport_name, client_config in service_transport.clients.items():
print(f" Transport: {transport_name}")
print(f" Client: {client_config.library_client}")
for rpc_name, method_list in client_config.rpcs.items():
print(f" RPC: {rpc_name}")
for method in method_list.methods:
print(f" Method: {method}")from google.cloud.location.locations_pb2 import Location
def plan_multi_region_deployment(locations: list[Location],
service_requirements: dict) -> dict:
"""Plan multi-region deployment based on requirements."""
deployment_plan = {
"primary_regions": [],
"secondary_regions": [],
"backup_regions": []
}
# Group locations by region
regions = {}
for location in locations:
region = location.location_id.split('-')[0] + '-' + location.location_id.split('-')[1]
if region not in regions:
regions[region] = []
regions[region].append(location)
# Sort regions by preference
preferred_regions = service_requirements.get("preferred_regions", [])
# Select primary regions
for region in preferred_regions:
if region in regions and len(deployment_plan["primary_regions"]) < 2:
deployment_plan["primary_regions"].extend(regions[region][:1])
# Select secondary regions
remaining_regions = [r for r in regions.keys()
if r not in [loc.location_id for loc in deployment_plan["primary_regions"]]]
for region in remaining_regions[:2]:
deployment_plan["secondary_regions"].extend(regions[region][:1])
# Select backup regions
for region in remaining_regions[2:4]:
deployment_plan["backup_regions"].extend(regions[region][:1])
return deployment_plan
# Example usage
requirements = {
"preferred_regions": ["us-central", "us-east"],
"min_regions": 3,
"disaster_recovery": True
}
deployment = plan_multi_region_deployment(locations, requirements)
print("Deployment Plan:")
print(f"Primary: {[loc.location_id for loc in deployment['primary_regions']]}")
print(f"Secondary: {[loc.location_id for loc in deployment['secondary_regions']]}")
print(f"Backup: {[loc.location_id for loc in deployment['backup_regions']]}")from google.cloud.location.locations_pb2 import Location
from google.protobuf.struct_pb2 import Struct
def extract_location_capabilities(location: Location) -> dict:
"""Extract service capabilities from location metadata."""
capabilities = {
"location_id": location.location_id,
"display_name": location.display_name,
"services": [],
"features": {},
"zones": []
}
# Extract from labels
for key, value in location.labels.items():
if key.startswith("service/"):
service_name = key.replace("service/", "")
if value.lower() == "true":
capabilities["services"].append(service_name)
elif key.startswith("feature/"):
feature_name = key.replace("feature/", "")
capabilities["features"][feature_name] = value
elif key == "zones":
capabilities["zones"] = value.split(",")
# Extract from metadata if available
if location.metadata:
# Metadata processing would be service-specific
pass
return capabilities
# Example usage
for location in locations:
caps = extract_location_capabilities(location)
print(f"\nLocation: {caps['location_id']}")
print(f"Services: {', '.join(caps['services'])}")
print(f"Zones: {', '.join(caps['zones'])}")
if caps['features']:
print("Features:")
for feature, value in caps['features'].items():
print(f" {feature}: {value}")def calculate_deployment_cost(locations: list[Location],
usage_requirements: dict) -> dict:
"""Calculate estimated deployment costs by location."""
cost_estimates = {}
# Sample cost factors (would come from actual pricing data)
base_costs = {
"us-central": 1.0, # Base cost multiplier
"us-east": 0.95, # 5% cheaper
"us-west": 1.1, # 10% more expensive
"europe-west": 1.2, # 20% more expensive
"asia-east": 1.15 # 15% more expensive
}
for location in locations:
region_prefix = '-'.join(location.location_id.split('-')[:2])
base_multiplier = base_costs.get(region_prefix, 1.0)
# Calculate costs based on usage
compute_cost = usage_requirements.get("compute_hours", 0) * base_multiplier * 0.10
storage_cost = usage_requirements.get("storage_gb", 0) * base_multiplier * 0.02
network_cost = usage_requirements.get("network_gb", 0) * base_multiplier * 0.08
total_cost = compute_cost + storage_cost + network_cost
cost_estimates[location.location_id] = {
"compute": compute_cost,
"storage": storage_cost,
"network": network_cost,
"total": total_cost,
"multiplier": base_multiplier
}
return cost_estimates
# Example usage
usage = {
"compute_hours": 720, # 1 month
"storage_gb": 1000, # 1TB
"network_gb": 500 # 500GB transfer
}
costs = calculate_deployment_cost(locations, usage)
for location_id, cost_info in sorted(costs.items(), key=lambda x: x[1]["total"]):
print(f"{location_id}: ${cost_info['total']:.2f}/month")Install with Tessl CLI
npx tessl i tessl/pypi-googleapis-common-protos