tessl/pypi-azure-mgmt-msi

Microsoft Azure Managed Service Identity 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

User-Assigned Identity Management

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

Comprehensive lifecycle management for user-assigned managed identities in Azure. User-assigned identities are standalone Azure resources that can be assigned to multiple Azure resources, providing a consistent identity that can be shared across virtual machines, app services, and other Azure services.

Capabilities

Identity Creation

Creates or updates a user-assigned managed identity in a specified resource group. This operation is idempotent - calling it multiple times with the same parameters will not cause errors.

def create_or_update(
    resource_group_name: str,
    resource_name: str,
    parameters: Union[Identity, IO[bytes]],
    **kwargs
) -> Identity:
    """
    Create or update a user-assigned identity.

    Parameters:
    - resource_group_name (str): Name of the resource group
    - resource_name (str): Name of the identity resource
    - parameters (Union[Identity, IO[bytes]]): Identity creation parameters or raw JSON bytes
    - **kwargs: Additional request options

    Returns:
    Identity: The created or updated identity resource

    Raises:
    ResourceExistsError: If identity exists with different configuration
    HttpResponseError: For other API errors
    """

Usage Example:

from azure.mgmt.msi import ManagedServiceIdentityClient
from azure.identity import DefaultAzureCredential

client = ManagedServiceIdentityClient(
    credential=DefaultAzureCredential(),
    subscription_id="your-subscription-id"
)

# Create a user-assigned identity
identity = client.user_assigned_identities.create_or_update(
    resource_group_name="myResourceGroup",
    resource_name="myUserIdentity",
    parameters={
        "location": "eastus",
        "tags": {
            "environment": "production",
            "team": "platform",
            "project": "identity-management"
        }
    }
)

print(f"Created identity: {identity.name}")
print(f"Client ID: {identity.client_id}")
print(f"Principal ID: {identity.principal_id}")
print(f"Tenant ID: {identity.tenant_id}")

Identity Retrieval

Retrieves an existing user-assigned managed identity by name within a resource group.

def get(
    resource_group_name: str,
    resource_name: str,
    **kwargs
) -> Identity:
    """
    Get a user-assigned identity.

    Parameters:
    - resource_group_name (str): Name of the resource group
    - resource_name (str): Name of the identity resource
    - **kwargs: Additional request options

    Returns:
    Identity: The requested identity resource

    Raises:
    ResourceNotFoundError: If identity does not exist
    HttpResponseError: For other API errors
    """

Usage Example:

try:
    identity = client.user_assigned_identities.get(
        resource_group_name="myResourceGroup",
        resource_name="myUserIdentity"
    )
    
    print(f"Identity found: {identity.name}")
    print(f"Location: {identity.location}")
    print(f"Tags: {identity.tags}")
    print(f"Client ID: {identity.client_id}")
    
except ResourceNotFoundError:
    print("Identity not found")

Identity Updates

Updates tags and other mutable properties of an existing user-assigned managed identity. Note that core identity properties like client_id and principal_id cannot be modified.

def update(
    resource_group_name: str,
    resource_name: str,
    parameters: Union[IdentityUpdate, IO[bytes]],
    **kwargs
) -> Identity:
    """
    Update a user-assigned identity.

    Parameters:
    - resource_group_name (str): Name of the resource group
    - resource_name (str): Name of the identity resource
    - parameters (Union[IdentityUpdate, IO[bytes]]): Update parameters or raw JSON bytes
    - **kwargs: Additional request options

    Returns:
    Identity: The updated identity resource

    Raises:
    ResourceNotFoundError: If identity does not exist
    HttpResponseError: For other API errors
    """

Usage Example:

# Update tags and location (if supported)
updated_identity = client.user_assigned_identities.update(
    resource_group_name="myResourceGroup",
    resource_name="myUserIdentity",
    parameters={
        "tags": {
            "environment": "staging",
            "updated": "2024-01-15",
            "team": "platform"
        }
    }
)

print(f"Updated identity tags: {updated_identity.tags}")

Identity Deletion

Permanently deletes a user-assigned managed identity. This operation cannot be undone and will fail if the identity is currently assigned to any Azure resources.

def delete(
    resource_group_name: str,
    resource_name: str,
    **kwargs
) -> None:
    """
    Delete a user-assigned identity.

    Parameters:
    - resource_group_name (str): Name of the resource group
    - resource_name (str): Name of the identity resource
    - **kwargs: Additional request options

    Returns:
    None

    Raises:
    ResourceNotFoundError: If identity does not exist
    HttpResponseError: If identity is still in use or other API errors
    """

Usage Example:

try:
    client.user_assigned_identities.delete(
        resource_group_name="myResourceGroup",
        resource_name="myUserIdentity"
    )
    print("Identity deleted successfully")
    
except HttpResponseError as e:
    if e.status_code == 409:
        print("Cannot delete identity - it is still assigned to resources")
    else:
        print(f"Deletion failed: {e.message}")

Subscription-Level Listing

Lists all user-assigned identities within an Azure subscription, with automatic pagination support.

def list_by_subscription(**kwargs) -> ItemPaged[Identity]:
    """
    List user-assigned identities in a subscription.

    Parameters:
    - **kwargs: Additional request options

    Returns:
    ItemPaged[Identity]: Paginated list of identity resources

    Raises:
    HttpResponseError: For API errors
    """

Usage Example:

# List all identities in subscription
print("User-assigned identities in subscription:")
for identity in client.user_assigned_identities.list_by_subscription():
    print(f"- {identity.name} (Resource Group: {identity.id.split('/')[4]})")
    print(f"  Location: {identity.location}")
    print(f"  Client ID: {identity.client_id}")
    print(f"  Tags: {identity.tags}")

# Count identities
identity_count = len(list(client.user_assigned_identities.list_by_subscription()))
print(f"Total identities: {identity_count}")

Resource Group Listing

Lists all user-assigned identities within a specific resource group, with automatic pagination support.

def list_by_resource_group(
    resource_group_name: str,
    **kwargs
) -> ItemPaged[Identity]:
    """
    List user-assigned identities in a resource group.

    Parameters:
    - resource_group_name (str): Name of the resource group
    - **kwargs: Additional request options

    Returns:
    ItemPaged[Identity]: Paginated list of identity resources

    Raises:
    HttpResponseError: For API errors
    """

Usage Example:

# List identities in specific resource group
resource_group = "myResourceGroup"
print(f"Identities in resource group '{resource_group}':")

for identity in client.user_assigned_identities.list_by_resource_group(
    resource_group_name=resource_group
):
    print(f"- {identity.name}")
    print(f"  Client ID: {identity.client_id}")
    print(f"  Principal ID: {identity.principal_id}")
    
    # Check if identity has federated credentials
    fed_creds = list(client.federated_identity_credentials.list(
        resource_group_name=resource_group,
        resource_name=identity.name
    ))
    
    if fed_creds:
        print(f"  Federated credentials: {len(fed_creds)}")

Advanced Usage Patterns

Bulk Identity Management

def create_identities_for_environments(resource_group: str, base_name: str, environments: list):
    """Create user-assigned identities for multiple environments."""
    created_identities = {}
    
    for env in environments:
        identity_name = f"{base_name}-{env}"
        
        try:
            identity = client.user_assigned_identities.create_or_update(
                resource_group_name=resource_group,
                resource_name=identity_name,
                parameters={
                    "location": "eastus",
                    "tags": {
                        "environment": env,
                        "managed-by": "automation",
                        "base-name": base_name
                    }
                }
            )
            
            created_identities[env] = {
                "name": identity.name,
                "client_id": identity.client_id,
                "principal_id": identity.principal_id
            }
            
        except Exception as e:
            print(f"Failed to create identity for {env}: {e}")
    
    return created_identities

# Create identities for dev, staging, and prod
environments = ["dev", "staging", "prod"]
identities = create_identities_for_environments(
    resource_group="myResourceGroup",
    base_name="myapp-identity",
    environments=environments
)

Identity Validation and Health Checks

def validate_identity_health(resource_group: str, identity_name: str):
    """Validate that an identity exists and is properly configured."""
    try:
        identity = client.user_assigned_identities.get(
            resource_group_name=resource_group,
            resource_name=identity_name
        )
        
        # Basic validation
        health_check = {
            "exists": True,
            "name": identity.name,
            "client_id": identity.client_id,
            "principal_id": identity.principal_id,
            "tenant_id": identity.tenant_id,
            "location": identity.location,
            "has_tags": bool(identity.tags),
            "federated_credentials": []
        }
        
        # Check for federated credentials
        try:
            fed_creds = list(client.federated_identity_credentials.list(
                resource_group_name=resource_group,
                resource_name=identity_name
            ))
            health_check["federated_credentials"] = [
                {
                    "name": cred.name,
                    "issuer": cred.issuer,
                    "subject": cred.subject,
                    "audiences": cred.audiences
                }
                for cred in fed_creds
            ]
        except Exception as e:
            health_check["federated_credentials_error"] = str(e)
        
        return health_check
        
    except ResourceNotFoundError:
        return {"exists": False, "error": "Identity not found"}
    except Exception as e:
        return {"exists": False, "error": str(e)}

# Validate identity
health = validate_identity_health("myResourceGroup", "myUserIdentity")
print(f"Identity health: {health}")

Types

class Identity:
    """User-assigned managed identity resource."""
    # Read-only properties
    id: str                    # Full Azure resource ID
    name: str                  # Resource name
    type: str                  # Resource type (Microsoft.ManagedIdentity/userAssignedIdentities)
    system_data: SystemData    # ARM metadata (creation/modification info)
    tenant_id: str            # Azure tenant ID where identity is created
    principal_id: str         # Service principal object ID in Azure AD
    client_id: str            # Application (client) ID for the identity
    
    # Configurable properties
    location: str             # Azure region (required for creation)
    tags: Dict[str, str]      # Resource tags (optional)
    isolation_scope: IsolationScope # Regional isolation setting (None or Regional)

class IdentityUpdate:
    """Update parameters for user-assigned identity."""
    # Read-only properties
    id: str                         # Resource ID (read-only)
    name: str                       # Resource name (read-only)
    type: str                       # Resource type (read-only)
    system_data: SystemData         # ARM metadata (read-only)
    tenant_id: str                  # Azure tenant ID (read-only)
    principal_id: str               # Service principal ID (read-only)
    client_id: str                  # Application client ID (read-only)
    
    # Configurable properties
    location: str                   # Updated location (optional)
    tags: Dict[str, str]            # Updated resource tags (optional)
    isolation_scope: IsolationScope # Regional isolation setting (optional)

Install with Tessl CLI