Microsoft Azure ComputeSchedule Management Client Library for Python providing VM scheduling operations.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Monitor operation status, cancel running operations, and retrieve detailed error information for troubleshooting failed operations. These capabilities provide comprehensive lifecycle management for all submitted operations.
Retrieve current status information for operations using their operation IDs, including execution state and progress details.
def virtual_machines_get_operation_status(
self,
locationparameter: str,
request_body: Union[GetOperationStatusRequest, JSON, IO[bytes]],
**kwargs: Any
) -> GetOperationStatusResponse:
"""
Get status information for operations by operation IDs.
Parameters:
- locationparameter: Azure region where operations are executing (required)
- request_body: Request containing operation IDs to query (required)
Returns:
Response containing status information for each requested operation
"""Usage Example:
from azure.mgmt.computeschedule.models import GetOperationStatusRequest
# Query status for multiple operations
request = GetOperationStatusRequest(
operation_ids=[
"op-12345678-abcd-1234-efgh-123456789012",
"op-87654321-dcba-4321-hgfe-210987654321"
]
)
response = client.scheduled_actions.virtual_machines_get_operation_status(
locationparameter="eastus",
request_body=request
)
# Check status of each operation
for operation in response.results:
print(f"Operation {operation.operation_id}: {operation.state}")
if operation.state == "SUCCEEDED":
print(f" Completed at: {operation.completion_time}")
elif operation.state == "FAILED":
print(f" Failed with error: {operation.error_message}")
elif operation.state == "EXECUTING":
print(f" Progress: {operation.progress_percentage}%")Cancel running or pending operations before they complete execution.
def virtual_machines_cancel_operations(
self,
locationparameter: str,
request_body: Union[CancelOperationsRequest, JSON, IO[bytes]],
**kwargs: Any
) -> CancelOperationsResponse:
"""
Cancel running operations using operation IDs.
Parameters:
- locationparameter: Azure region where operations are executing (required)
- request_body: Request containing operation IDs to cancel (required)
Returns:
Response indicating cancellation status for each operation
"""Usage Example:
from azure.mgmt.computeschedule.models import CancelOperationsRequest
# Cancel multiple operations
request = CancelOperationsRequest(
operation_ids=[
"op-12345678-abcd-1234-efgh-123456789012",
"op-87654321-dcba-4321-hgfe-210987654321"
]
)
response = client.scheduled_actions.virtual_machines_cancel_operations(
locationparameter="eastus",
request_body=request
)
# Check cancellation results
for result in response.results:
if result.cancelled:
print(f"Operation {result.operation_id} cancelled successfully")
else:
print(f"Operation {result.operation_id} could not be cancelled: {result.reason}")Retrieve detailed error information for failed operations to aid in troubleshooting and debugging.
def virtual_machines_get_operation_errors(
self,
locationparameter: str,
request_body: Union[GetOperationErrorsRequest, JSON, IO[bytes]],
**kwargs: Any
) -> GetOperationErrorsResponse:
"""
Get detailed error information for operations by operation IDs.
Parameters:
- locationparameter: Azure region where operations executed (required)
- request_body: Request containing operation IDs to query errors (required)
Returns:
Response containing detailed error information for each operation
"""Usage Example:
from azure.mgmt.computeschedule.models import GetOperationErrorsRequest
# Get error details for failed operations
request = GetOperationErrorsRequest(
operation_ids=[
"op-failed-1234-abcd-1234-efgh-123456789012",
"op-failed-5678-dcba-4321-hgfe-210987654321"
]
)
response = client.scheduled_actions.virtual_machines_get_operation_errors(
locationparameter="eastus",
request_body=request
)
# Process error information
for error_result in response.results:
print(f"Operation {error_result.operation_id} errors:")
for error in error_result.errors:
print(f" Error Code: {error.error_code}")
print(f" Error Message: {error.error_message}")
print(f" Target Resource: {error.target_resource}")
if error.additional_info:
print(f" Additional Info: {error.additional_info}")Retrieve the list of available operations supported by the ComputeSchedule provider.
def list(self, **kwargs: Any) -> ItemPaged[Operation]:
"""
List all available operations for the ComputeSchedule provider.
Returns:
Paginated list of Operation objects describing available operations
"""Usage Example:
# List all available provider operations
operations = client.operations.list()
print("Available ComputeSchedule operations:")
for operation in operations:
print(f" {operation.name}: {operation.display.description}")
print(f" Provider: {operation.display.provider}")
print(f" Resource: {operation.display.resource}")
print(f" Operation: {operation.display.operation}")Operations progress through several states during their lifecycle:
Monitor operation progress with polling:
import time
from azure.mgmt.computeschedule.models import GetOperationStatusRequest, OperationState
def wait_for_operation_completion(client, location, operation_id, timeout_minutes=30):
"""Wait for operation to complete with status polling."""
request = GetOperationStatusRequest(operation_ids=[operation_id])
timeout_seconds = timeout_minutes * 60
start_time = time.time()
while time.time() - start_time < timeout_seconds:
response = client.scheduled_actions.virtual_machines_get_operation_status(
locationparameter=location,
request_body=request
)
if response.results:
status = response.results[0]
print(f"Operation {operation_id} status: {status.state}")
if status.state in [OperationState.SUCCEEDED, OperationState.FAILED, OperationState.CANCELLED]:
return status
time.sleep(30) # Poll every 30 seconds
raise TimeoutError(f"Operation {operation_id} did not complete within {timeout_minutes} minutes")
# Usage
try:
final_status = wait_for_operation_completion(client, "eastus", operation_id)
if final_status.state == OperationState.SUCCEEDED:
print("Operation completed successfully!")
else:
print(f"Operation ended with status: {final_status.state}")
except TimeoutError as e:
print(f"Timeout: {e}")Handle multiple operations efficiently:
from azure.mgmt.computeschedule.models import (
GetOperationStatusRequest,
CancelOperationsRequest,
OperationState
)
def manage_batch_operations(client, location, operation_ids):
"""Monitor and manage multiple operations."""
# Get status for all operations
status_request = GetOperationStatusRequest(operation_ids=operation_ids)
status_response = client.scheduled_actions.virtual_machines_get_operation_status(
locationparameter=location,
request_body=status_request
)
# Separate operations by status
running_ops = []
failed_ops = []
completed_ops = []
for op_status in status_response.results:
if op_status.state in [OperationState.EXECUTING, OperationState.PENDING_EXECUTION]:
running_ops.append(op_status.operation_id)
elif op_status.state == OperationState.FAILED:
failed_ops.append(op_status.operation_id)
elif op_status.state == OperationState.SUCCEEDED:
completed_ops.append(op_status.operation_id)
print(f"Running: {len(running_ops)}, Failed: {len(failed_ops)}, Completed: {len(completed_ops)}")
# Cancel all running operations if needed
if running_ops:
cancel_request = CancelOperationsRequest(operation_ids=running_ops)
cancel_response = client.scheduled_actions.virtual_machines_cancel_operations(
locationparameter=location,
request_body=cancel_request
)
print(f"Cancelled {sum(1 for r in cancel_response.results if r.cancelled)} operations")
return {
'running': running_ops,
'failed': failed_ops,
'completed': completed_ops
}Analyze operation failures:
from azure.mgmt.computeschedule.models import GetOperationErrorsRequest
def analyze_operation_failures(client, location, failed_operation_ids):
"""Analyze detailed error information for failed operations."""
if not failed_operation_ids:
return
error_request = GetOperationErrorsRequest(operation_ids=failed_operation_ids)
error_response = client.scheduled_actions.virtual_machines_get_operation_errors(
locationparameter=location,
request_body=error_request
)
error_summary = {}
for op_errors in error_response.results:
op_id = op_errors.operation_id
error_summary[op_id] = []
for error in op_errors.errors:
error_info = {
'code': error.error_code,
'message': error.error_message,
'target': error.target_resource,
'additional_info': error.additional_info
}
error_summary[op_id].append(error_info)
# Print error summary
for op_id, errors in error_summary.items():
print(f"\nOperation {op_id} failures:")
for i, error in enumerate(errors, 1):
print(f" Error {i}:")
print(f" Code: {error['code']}")
print(f" Message: {error['message']}")
if error['target']:
print(f" Target: {error['target']}")
return error_summaryInstall with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-computeschedule