tessl/pypi-azure-mgmt-machinelearningcompute

Microsoft Azure Machine Learning Compute Management Client Library for Python

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Cluster Operations

Name: tessl/pypi-azure-mgmt-machinelearningcompute
Author: tessl

Complete lifecycle management of operationalization clusters including creation, updates, deletion, credential management, and system services. These operations enable full control over Azure Machine Learning compute resources.

Capabilities

Cluster Lifecycle Management

Core operations for creating, retrieving, updating, and deleting operationalization clusters.

class OperationalizationClustersOperations:
    """Operations for managing operationalization clusters."""
    
    def create_or_update(
        self, 
        resource_group_name: str, 
        cluster_name: str, 
        parameters: OperationalizationCluster,
        custom_headers: Dict[str, str] = None,
        raw: bool = False
    ) -> AzureOperationPoller[OperationalizationCluster]:
        """
        Create or update an operationalization cluster.
        
        Args:
            resource_group_name (str): Name of the resource group in which the cluster is located
            cluster_name (str): The name of the cluster
            parameters (OperationalizationCluster): Parameters supplied to create or update cluster
            custom_headers (dict, optional): Headers to add to the request
            raw (bool): Returns direct response alongside deserialized response
            
        Returns:
            AzureOperationPoller that returns OperationalizationCluster when complete
            
        Raises:
            ErrorResponseWrapperException: Service error occurred
        """
    
    def get(
        self,
        resource_group_name: str,
        cluster_name: str,
        custom_headers: Dict[str, str] = None,
        raw: bool = False
    ) -> OperationalizationCluster:
        """
        Gets the operationalization cluster resource view.
        Note: credentials are not returned by this call, use list_keys to get them.
        
        Args:
            resource_group_name (str): Name of the resource group in which the cluster is located
            cluster_name (str): The name of the cluster
            custom_headers (dict, optional): Headers to add to the request
            raw (bool): Returns direct response alongside deserialized response
            
        Returns:
            OperationalizationCluster: The cluster resource
            
        Raises:
            ErrorResponseWrapperException: Service error occurred
        """
    
    def update(
        self,
        resource_group_name: str,
        cluster_name: str,
        tags: Dict[str, str] = None,
        custom_headers: Dict[str, str] = None,
        raw: bool = False
    ) -> OperationalizationCluster:
        """
        Update cluster tags. Use PUT operation to update other properties.
        
        Args:
            resource_group_name (str): Name of the resource group in which the cluster is located
            cluster_name (str): The name of the cluster
            tags (dict, optional): Resource tags (maximum 15 tags, key ≤128 chars, value ≤256 chars)
            custom_headers (dict, optional): Headers to add to the request
            raw (bool): Returns direct response alongside deserialized response
            
        Returns:
            OperationalizationCluster: The updated cluster resource
            
        Raises:
            ErrorResponseWrapperException: Service error occurred
        """
    
    def delete(
        self,
        resource_group_name: str,
        cluster_name: str,
        delete_all: bool = None,
        custom_headers: Dict[str, str] = None,
        raw: bool = False
    ) -> AzureOperationPoller[None]:
        """
        Deletes the specified cluster.
        
        Args:
            resource_group_name (str): Name of the resource group in which the cluster is located
            cluster_name (str): The name of the cluster
            delete_all (bool, optional): If true, deletes all resources associated with this cluster
            custom_headers (dict, optional): Headers to add to the request
            raw (bool): Returns direct response alongside deserialized response
            
        Returns:
            AzureOperationPoller that returns None when complete
            
        Raises:
            ErrorResponseWrapperException: Service error occurred
        """

Usage Examples:

from azure.mgmt.machinelearningcompute.models import (
    OperationalizationCluster,
    ClusterType,
    StorageAccountProperties,
    AcsClusterProperties
)

# Create a simple local cluster
local_cluster = OperationalizationCluster(
    location="eastus",
    cluster_type=ClusterType.local,
    description="Development cluster",
    storage_account=StorageAccountProperties(
        resource_id="/subscriptions/.../storageAccounts/devstorage"
    )
)

# Create the cluster (long-running operation)
create_operation = client.operationalization_clusters.create_or_update(
    resource_group_name="dev-rg",
    cluster_name="dev-cluster",
    parameters=local_cluster
)
cluster = create_operation.result()  # Wait for completion

# Get cluster details
cluster = client.operationalization_clusters.get("dev-rg", "dev-cluster")
print(f"Cluster status: {cluster.provisioning_state}")

# Update cluster tags
updated_cluster = client.operationalization_clusters.update(
    resource_group_name="dev-rg",
    cluster_name="dev-cluster",
    tags={"environment": "development", "owner": "ml-team"}
)

# Delete cluster and all associated resources
delete_operation = client.operationalization_clusters.delete(
    resource_group_name="dev-rg",
    cluster_name="dev-cluster",
    delete_all=True
)
delete_operation.wait()  # Wait for completion

Cluster Listing Operations

Operations for discovering clusters across resource groups and subscriptions.

def list_by_resource_group(
    self,
    resource_group_name: str,
    skiptoken: str = None,
    custom_headers: Dict[str, str] = None,
    raw: bool = False
) -> OperationalizationClusterPaged:
    """
    Gets the clusters in the specified resource group.
    
    Args:
        resource_group_name (str): Name of the resource group in which the cluster is located
        skiptoken (str, optional): Continuation token for pagination
        custom_headers (dict, optional): Headers to add to the request
        raw (bool): Returns direct response alongside deserialized response
        
    Returns:
        OperationalizationClusterPaged: Paginated collection of clusters
        
    Raises:
        CloudError: Service error occurred
    """

def list_by_subscription_id(
    self,
    skiptoken: str = None,
    custom_headers: Dict[str, str] = None,
    raw: bool = False
) -> OperationalizationClusterPaged:
    """
    Gets the operationalization clusters in the specified subscription.
    
    Args:
        skiptoken (str, optional): Continuation token for pagination
        custom_headers (dict, optional): Headers to add to the request
        raw (bool): Returns direct response alongside deserialized response
        
    Returns:
        OperationalizationClusterPaged: Paginated collection of clusters
        
    Raises:
        CloudError: Service error occurred
    """

Usage Examples:

# List all clusters in a resource group
clusters = client.operationalization_clusters.list_by_resource_group("prod-rg")
for cluster in clusters:
    print(f"Cluster: {cluster.name}, Status: {cluster.provisioning_state}")

# List all clusters in the subscription
all_clusters = client.operationalization_clusters.list_by_subscription_id()
for cluster in all_clusters:
    print(f"Cluster: {cluster.name} in RG: {cluster.id.split('/')[4]}")

# Handle pagination manually if needed
clusters_page = client.operationalization_clusters.list_by_resource_group("large-rg")
first_10 = list(clusters_page)[:10]

Credential Management

Operations for retrieving cluster access credentials for storage, container registry, and container service components.

def list_keys(
    self,
    resource_group_name: str,
    cluster_name: str,
    custom_headers: Dict[str, str] = None,
    raw: bool = False
) -> OperationalizationClusterCredentials:
    """
    Gets the credentials for the specified cluster such as Storage, ACR and ACS credentials.
    This is a long running operation because it fetches keys from dependencies.
    
    Args:
        resource_group_name (str): Name of the resource group in which the cluster is located
        cluster_name (str): The name of the cluster
        custom_headers (dict, optional): Headers to add to the request
        raw (bool): Returns direct response alongside deserialized response
        
    Returns:
        OperationalizationClusterCredentials: Cluster access credentials
        
    Raises:
        CloudError: Service error occurred
    """

Usage Example:

# Get cluster credentials
credentials = client.operationalization_clusters.list_keys("prod-rg", "prod-cluster")

# Access different credential types
if credentials.storage_account:
    print(f"Storage account key: {credentials.storage_account.primary_key}")

if credentials.container_registry:
    print(f"Registry username: {credentials.container_registry.username}")

if credentials.container_service:
    print(f"Kube config available: {bool(credentials.container_service.kube_config)}")

System Services Management

Operations for managing system services within clusters, including update checking and service updates.

def check_system_services_updates_available(
    self,
    resource_group_name: str,
    cluster_name: str,
    custom_headers: Dict[str, str] = None,
    raw: bool = False
) -> CheckSystemServicesUpdatesAvailableResponse:
    """
    Checks if updates are available for system services in the cluster.
    
    Args:
        resource_group_name (str): Name of the resource group in which the cluster is located
        cluster_name (str): The name of the cluster
        custom_headers (dict, optional): Headers to add to the request
        raw (bool): Returns direct response alongside deserialized response
        
    Returns:
        CheckSystemServicesUpdatesAvailableResponse: Update availability information
        
    Raises:
        CloudError: Service error occurred
    """

def update_system_services(
    self,
    resource_group_name: str,
    cluster_name: str,
    custom_headers: Dict[str, str] = None,
    raw: bool = False
) -> AzureOperationPoller[UpdateSystemServicesResponse]:
    """
    Updates system services in a cluster.
    
    Args:
        resource_group_name (str): Name of the resource group in which the cluster is located
        cluster_name (str): The name of the cluster
        custom_headers (dict, optional): Headers to add to the request
        raw (bool): Returns direct response alongside deserialized response
        
    Returns:
        AzureOperationPoller that returns UpdateSystemServicesResponse when complete
        
    Raises:
        CloudError: Service error occurred
    """

Usage Examples:

# Check for system service updates
update_check = client.operationalization_clusters.check_system_services_updates_available(
    "prod-rg", "prod-cluster"
)

if update_check.updates_available == "Yes":
    print("Updates are available")
    
    # Apply system service updates (long-running operation)
    update_operation = client.operationalization_clusters.update_system_services(
        "prod-rg", "prod-cluster"
    )
    
    # Wait for update completion
    update_result = update_operation.result()
    print(f"Update status: {update_result.update_status}")
else:
    print("No updates available")

Long-Running Operations

Many cluster operations are long-running and return AzureOperationPoller objects:

# Start a long-running operation
operation = client.operationalization_clusters.create_or_update(rg, name, params)

# Wait for completion (blocking)
result = operation.result()

# Check if operation is done (non-blocking)
if operation.done():
    result = operation.result()
else:
    print("Still in progress...")

# Wait with timeout
try:
    result = operation.result(timeout=600)  # 10 minutes
except Exception as e:
    print(f"Operation timed out: {e}")

Install with Tessl CLI

npx tessl i tessl/pypi-azure-mgmt-machinelearningcompute

docs

client-management.md

cluster-operations.md

tessl/pypi-azure-mgmt-machinelearningcompute