Management of service revisions and their lifecycle. Revisions are immutable snapshots of service configurations that enable controlled deployments, traffic splitting, and rollbacks.
Create and configure revision clients for managing service deployments.
class RevisionsClient:
"""Synchronous client for managing Cloud Run revisions."""
def __init__(self, *, credentials=None, transport=None, client_options=None, client_info=None):
"""
Initialize the Revisions client.
Args:
credentials: Optional authentication credentials
transport: Transport to use for requests (grpc, grpc_asyncio, rest)
client_options: Client configuration options
client_info: Client information for user agent
"""
class RevisionsAsyncClient:
"""Asynchronous client for managing Cloud Run revisions."""Get revision details, configuration, and status information.
def get_revision(
self,
request: GetRevisionRequest = None,
*,
name: str = None,
**kwargs
) -> Revision:
"""
Get a Cloud Run revision.
Args:
request: The request object
name: Required. The name of the revision. Format: projects/{project}/locations/{location}/services/{service}/revisions/{revision}
Returns:
Revision: The revision configuration and status
"""List revisions for a service with filtering and pagination.
def list_revisions(
self,
request: ListRevisionsRequest = None,
*,
parent: str = None,
**kwargs
) -> ListRevisionsResponse:
"""
List Cloud Run revisions for a service.
Args:
request: The request object
parent: Required. The service to list revisions for. Format: projects/{project}/locations/{location}/services/{service}
Returns:
ListRevisionsResponse: Paginated list of revisions
"""Delete unused revisions to clean up resources.
def delete_revision(
self,
request: DeleteRevisionRequest = None,
*,
name: str = None,
**kwargs
) -> Operation:
"""
Delete a Cloud Run revision.
Args:
request: The request object
name: Required. The name of the revision to delete
Returns:
Operation: Long-running operation for the deletion
"""class Revision:
"""
A Cloud Run revision configuration.
Attributes:
name (str): The unique name of the revision
uid (str): Unique identifier assigned by the system
generation (int): A sequence number representing a specific generation
labels (dict): User-defined labels
annotations (dict): User-defined annotations
create_time (Timestamp): The creation time
update_time (Timestamp): The last modification time
delete_time (Timestamp): The deletion time
expire_time (Timestamp): When the revision expires
launch_stage (LaunchStage): The launch stage of the revision
service (str): The name of the service that owns this revision
scaling (RevisionScaling): Scaling settings for this revision
vpc_access (VpcAccess): VPC access configuration
max_instance_request_concurrency (int): Maximum requests per instance
timeout (str): Maximum duration for request processing
service_account (str): Email address of the IAM service account
containers (list[Container]): The containers that define the revision
volumes (list[Volume]): Volumes to make available to containers
execution_environment (ExecutionEnvironment): The execution environment
encryption_key (str): Reference to a customer managed encryption key
reconciling (bool): Whether the revision is currently being reconciled
conditions (list[Condition]): Detailed status conditions
observed_generation (int): The generation observed by the controller
log_uri (str): URI where logs for this revision can be found
satisfies_pzs (bool): Whether the revision satisfies PZS requirements
session_affinity (bool): Whether sessions are routed to the same revision
scaling_status (RevisionScalingStatus): Current scaling status
etag (str): A fingerprint used for optimistic concurrency control
"""class GetRevisionRequest:
"""
Request message for getting a revision.
Attributes:
name (str): Required. The name of the revision to retrieve
"""
class ListRevisionsRequest:
"""
Request message for listing revisions.
Attributes:
parent (str): Required. The service to list revisions for
page_size (int): Maximum number of revisions to return
page_token (str): Token for retrieving the next page
show_deleted (bool): Whether to include deleted revisions
"""
class DeleteRevisionRequest:
"""
Request message for deleting a revision.
Attributes:
name (str): Required. The name of the revision to delete
validate_only (bool): Indicates whether to validate only
etag (str): A fingerprint for optimistic concurrency control
"""class ListRevisionsResponse:
"""
Response message for listing revisions.
Attributes:
revisions (list[Revision]): The list of revisions
next_page_token (str): Token for retrieving the next page
"""from google.cloud import run_v2
def split_traffic_between_revisions(service_name, revision_percentages):
"""
Split traffic between multiple revisions.
Args:
service_name: Full name of the service
revision_percentages: Dict mapping revision names to percentage
"""
services_client = run_v2.ServicesClient()
# Get current service configuration
service = services_client.get_service(name=service_name)
# Update traffic allocation
service.spec.traffic = []
for revision_name, percentage in revision_percentages.items():
traffic_target = run_v2.TrafficTarget(
type=run_v2.TrafficTargetAllocationType.TRAFFIC_TARGET_ALLOCATION_TYPE_REVISION,
revision=revision_name,
percent=percentage
)
service.spec.traffic.append(traffic_target)
# Apply the update
update_request = run_v2.UpdateServiceRequest(service=service)
operation = services_client.update_service(request=update_request)
updated_service = operation.result()
print(f"Traffic split updated for {service_name}")
return updated_service
# Usage
split_traffic_between_revisions(
"projects/my-project/locations/us-central1/services/my-app",
{
"my-app-rev-001": 80, # 80% to stable revision
"my-app-rev-002": 20 # 20% to new revision
}
)def cleanup_old_revisions(service_name, keep_count=5):
"""
Delete old revisions, keeping only the most recent ones.
Args:
service_name: Full name of the service
keep_count: Number of recent revisions to keep
"""
revisions_client = run_v2.RevisionsClient()
# List all revisions for the service
request = run_v2.ListRevisionsRequest(parent=service_name)
revisions = list(revisions_client.list_revisions(request=request))
# Sort by creation time (newest first)
revisions.sort(key=lambda r: r.create_time, reverse=True)
# Delete old revisions
for revision in revisions[keep_count:]:
print(f"Deleting old revision: {revision.name}")
delete_request = run_v2.DeleteRevisionRequest(name=revision.name)
operation = revisions_client.delete_revision(request=delete_request)
operation.result() # Wait for deletion
print(f"Kept {min(len(revisions), keep_count)} most recent revisions")def rollback_to_revision(service_name, target_revision_name):
"""
Rollback service to a specific revision.
Args:
service_name: Full name of the service
target_revision_name: Name of the revision to rollback to
"""
services_client = run_v2.ServicesClient()
# Get current service configuration
service = services_client.get_service(name=service_name)
# Set 100% traffic to target revision
service.spec.traffic = [
run_v2.TrafficTarget(
type=run_v2.TrafficTargetAllocationType.TRAFFIC_TARGET_ALLOCATION_TYPE_REVISION,
revision=target_revision_name,
percent=100
)
]
# Apply the rollback
update_request = run_v2.UpdateServiceRequest(service=service)
operation = services_client.update_service(request=update_request)
updated_service = operation.result()
print(f"Rolled back {service_name} to revision {target_revision_name}")
return updated_servicedef monitor_revision_health(service_name):
"""
Monitor the health status of all revisions for a service.
"""
revisions_client = run_v2.RevisionsClient()
request = run_v2.ListRevisionsRequest(parent=service_name)
healthy_revisions = []
unhealthy_revisions = []
for revision in revisions_client.list_revisions(request=request):
print(f"Revision: {revision.name}")
print(f" Created: {revision.create_time}")
print(f" Scaling: {revision.scaling_status}")
# Check conditions for health
ready = False
for condition in revision.conditions:
if condition.type == "Ready":
ready = condition.status == "True"
if not ready:
print(f" Status: {condition.reason} - {condition.message}")
break
if ready:
healthy_revisions.append(revision)
print(" Status: Healthy")
else:
unhealthy_revisions.append(revision)
print(" Status: Unhealthy")
print()
print(f"Summary: {len(healthy_revisions)} healthy, {len(unhealthy_revisions)} unhealthy")
return healthy_revisions, unhealthy_revisions