tessl/pypi-azure-mgmt-kusto

Microsoft Azure Kusto 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

Security Management

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

Comprehensive security management including cluster and database-level principal assignments, private endpoints, managed private endpoints, and private link resources. This enables fine-grained access control and network isolation for Kusto resources.

Capabilities

Cluster Principal Assignments

Management of security principals with cluster-level access permissions.

def get(
    resource_group_name: str,
    cluster_name: str,
    principal_assignment_name: str,
    **kwargs
) -> ClusterPrincipalAssignment:
    """
    Get a cluster principal assignment.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - principal_assignment_name: Name of the principal assignment
    
    Returns:
    ClusterPrincipalAssignment object
    """

def begin_create_or_update(
    resource_group_name: str,
    cluster_name: str,
    principal_assignment_name: str,
    parameters: ClusterPrincipalAssignment,
    **kwargs
) -> LROPoller[ClusterPrincipalAssignment]:
    """
    Create or update a cluster principal assignment.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - principal_assignment_name: Name of the principal assignment
    - parameters: ClusterPrincipalAssignment with configuration
    
    Returns:
    LROPoller for the long-running operation returning ClusterPrincipalAssignment
    """

def begin_delete(
    resource_group_name: str,
    cluster_name: str,
    principal_assignment_name: str,
    **kwargs
) -> LROPoller[None]:
    """
    Delete a cluster principal assignment.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - principal_assignment_name: Name of the principal assignment
    
    Returns:
    LROPoller for the long-running delete operation
    """

def list(
    resource_group_name: str,
    cluster_name: str,
    **kwargs
) -> Iterable[ClusterPrincipalAssignment]:
    """
    List cluster principal assignments.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    
    Returns:
    Iterable of ClusterPrincipalAssignment objects
    """

def check_name_availability(
    resource_group_name: str,
    cluster_name: str,
    principal_assignment_name: ClusterPrincipalAssignmentCheckNameRequest,
    **kwargs
) -> CheckNameResult:
    """
    Check if a cluster principal assignment name is available.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - principal_assignment_name: CheckNameRequest with name to validate
    
    Returns:
    CheckNameResult indicating availability
    """

Database Principal Assignments

Management of security principals with database-level access permissions.

def get(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    principal_assignment_name: str,
    **kwargs
) -> DatabasePrincipalAssignment:
    """
    Get a database principal assignment.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - principal_assignment_name: Name of the principal assignment
    
    Returns:
    DatabasePrincipalAssignment object
    """

def begin_create_or_update(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    principal_assignment_name: str,
    parameters: DatabasePrincipalAssignment,
    **kwargs
) -> LROPoller[DatabasePrincipalAssignment]:
    """
    Create or update a database principal assignment.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - principal_assignment_name: Name of the principal assignment
    - parameters: DatabasePrincipalAssignment with configuration
    
    Returns:
    LROPoller for the long-running operation returning DatabasePrincipalAssignment
    """

def begin_delete(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    principal_assignment_name: str,
    **kwargs
) -> LROPoller[None]:
    """
    Delete a database principal assignment.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - principal_assignment_name: Name of the principal assignment
    
    Returns:
    LROPoller for the long-running delete operation
    """

def list(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    **kwargs
) -> Iterable[DatabasePrincipalAssignment]:
    """
    List database principal assignments.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    
    Returns:
    Iterable of DatabasePrincipalAssignment objects
    """

def check_name_availability(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    principal_assignment_name: DatabasePrincipalAssignmentCheckNameRequest,
    **kwargs
) -> CheckNameResult:
    """
    Check if a database principal assignment name is available.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - principal_assignment_name: CheckNameRequest with name to validate
    
    Returns:
    CheckNameResult indicating availability
    """

Private Endpoint Connections

Management of private endpoint connections for secure network access to Kusto clusters.

def get(
    resource_group_name: str,
    cluster_name: str,
    private_endpoint_connection_name: str,
    **kwargs
) -> PrivateEndpointConnection:
    """
    Get a private endpoint connection.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - private_endpoint_connection_name: Name of the private endpoint connection
    
    Returns:
    PrivateEndpointConnection object
    """

def begin_create_or_update(
    resource_group_name: str,
    cluster_name: str,
    private_endpoint_connection_name: str,
    parameters: PrivateEndpointConnection,
    **kwargs
) -> LROPoller[PrivateEndpointConnection]:
    """
    Create or update a private endpoint connection.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - private_endpoint_connection_name: Name of the private endpoint connection
    - parameters: PrivateEndpointConnection with configuration
    
    Returns:
    LROPoller for the long-running operation returning PrivateEndpointConnection
    """

def begin_delete(
    resource_group_name: str,
    cluster_name: str,
    private_endpoint_connection_name: str,
    **kwargs
) -> LROPoller[None]:
    """
    Delete a private endpoint connection.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - private_endpoint_connection_name: Name of the private endpoint connection
    
    Returns:
    LROPoller for the long-running delete operation
    """

def list(
    resource_group_name: str,
    cluster_name: str,
    **kwargs
) -> Iterable[PrivateEndpointConnection]:
    """
    List private endpoint connections for a cluster.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    
    Returns:
    Iterable of PrivateEndpointConnection objects
    """

Managed Private Endpoints

Management of managed private endpoints created within the Kusto cluster's managed virtual network.

def get(
    resource_group_name: str,
    cluster_name: str,
    managed_private_endpoint_name: str,
    **kwargs
) -> ManagedPrivateEndpoint:
    """
    Get a managed private endpoint.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - managed_private_endpoint_name: Name of the managed private endpoint
    
    Returns:
    ManagedPrivateEndpoint object
    """

def begin_create_or_update(
    resource_group_name: str,
    cluster_name: str,
    managed_private_endpoint_name: str,
    parameters: ManagedPrivateEndpoint,
    **kwargs
) -> LROPoller[ManagedPrivateEndpoint]:
    """
    Create or update a managed private endpoint.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - managed_private_endpoint_name: Name of the managed private endpoint
    - parameters: ManagedPrivateEndpoint with configuration
    
    Returns:
    LROPoller for the long-running operation returning ManagedPrivateEndpoint
    """

def begin_update(
    resource_group_name: str,
    cluster_name: str,
    managed_private_endpoint_name: str,
    parameters: ManagedPrivateEndpoint,
    **kwargs
) -> LROPoller[ManagedPrivateEndpoint]:
    """
    Update a managed private endpoint.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - managed_private_endpoint_name: Name of the managed private endpoint
    - parameters: ManagedPrivateEndpoint with updates
    
    Returns:
    LROPoller for the long-running operation returning updated ManagedPrivateEndpoint
    """

def begin_delete(
    resource_group_name: str,
    cluster_name: str,
    managed_private_endpoint_name: str,
    **kwargs
) -> LROPoller[None]:
    """
    Delete a managed private endpoint.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - managed_private_endpoint_name: Name of the managed private endpoint
    
    Returns:
    LROPoller for the long-running delete operation
    """

def list(
    resource_group_name: str,
    cluster_name: str,
    **kwargs
) -> Iterable[ManagedPrivateEndpoint]:
    """
    List managed private endpoints for a cluster.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    
    Returns:
    Iterable of ManagedPrivateEndpoint objects
    """

def check_name_availability(
    resource_group_name: str,
    cluster_name: str,
    resource_name: ManagedPrivateEndpointsCheckNameRequest,
    **kwargs
) -> CheckNameResult:
    """
    Check if a managed private endpoint name is available.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - resource_name: CheckNameRequest with name to validate
    
    Returns:
    CheckNameResult indicating availability
    """

Private Link Resources

Operations to discover private link resources available for private endpoint connections.

def get(
    resource_group_name: str,
    cluster_name: str,
    private_link_resource_name: str,
    **kwargs
) -> PrivateLinkResource:
    """
    Get a private link resource.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - private_link_resource_name: Name of the private link resource
    
    Returns:
    PrivateLinkResource object
    """

def list(
    resource_group_name: str,
    cluster_name: str,
    **kwargs
) -> Iterable[PrivateLinkResource]:
    """
    List private link resources for a cluster.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    
    Returns:
    Iterable of PrivateLinkResource objects
    """

Usage Examples

Assigning Cluster Admin Role

from azure.mgmt.kusto.models import (
    ClusterPrincipalAssignment,
    ClusterPrincipalRole,
    PrincipalType
)

# Create cluster principal assignment
cluster_principal = ClusterPrincipalAssignment(
    principal_id="12345678-1234-1234-1234-123456789012",  # Azure AD Object ID
    role=ClusterPrincipalRole.ALL_DATABASES_ADMIN,
    tenant_id="87654321-4321-4321-4321-210987654321",     # Azure AD Tenant ID
    principal_type=PrincipalType.USER
)

# Assign the principal
poller = client.cluster_principal_assignments.begin_create_or_update(
    resource_group_name="my-resource-group",
    cluster_name="my-cluster",
    principal_assignment_name="admin-assignment",
    parameters=cluster_principal
)

assignment = poller.result()
print(f"Cluster principal assigned: {assignment.principal_id}")

Assigning Database Viewer Role

from azure.mgmt.kusto.models import (
    DatabasePrincipalAssignment,
    DatabasePrincipalRole,
    PrincipalType
)

# Create database principal assignment
database_principal = DatabasePrincipalAssignment(
    principal_id="group-object-id-here",
    role=DatabasePrincipalRole.VIEWER,
    tenant_id="87654321-4321-4321-4321-210987654321",
    principal_type=PrincipalType.GROUP
)

# Assign the principal to database
poller = client.database_principal_assignments.begin_create_or_update(
    resource_group_name="my-resource-group",
    cluster_name="my-cluster",
    database_name="my-database",
    principal_assignment_name="viewer-group-assignment",
    parameters=database_principal
)

assignment = poller.result()
print(f"Database principal assigned: {assignment.principal_id}")

Creating a Private Endpoint Connection

from azure.mgmt.kusto.models import (
    PrivateEndpointConnection,
    PrivateLinkServiceConnectionStateProperty
)

# Configure private endpoint connection
connection_state = PrivateLinkServiceConnectionStateProperty(
    status="Approved",
    description="Approved by administrator"
)

private_endpoint_conn = PrivateEndpointConnection(
    private_link_service_connection_state=connection_state
)

# Create private endpoint connection
poller = client.private_endpoint_connections.begin_create_or_update(
    resource_group_name="my-resource-group",
    cluster_name="my-cluster",
    private_endpoint_connection_name="my-private-endpoint",
    parameters=private_endpoint_conn
)

connection = poller.result()
print(f"Private endpoint connection created: {connection.name}")

Creating a Managed Private Endpoint

from azure.mgmt.kusto.models import ManagedPrivateEndpoint

# Configure managed private endpoint
managed_endpoint = ManagedPrivateEndpoint(
    private_link_resource_id="/subscriptions/sub-id/resourceGroups/rg/providers/Microsoft.Storage/storageAccounts/mystorageaccount",
    private_link_resource_region="East US",
    group_id="blob",
    request_message="Please approve this connection"
)

# Create managed private endpoint
poller = client.managed_private_endpoints.begin_create_or_update(
    resource_group_name="my-resource-group",
    cluster_name="my-cluster",
    managed_private_endpoint_name="storage-endpoint",
    parameters=managed_endpoint
)

endpoint = poller.result()
print(f"Managed private endpoint created: {endpoint.name}")

Key Types

class ClusterPrincipalAssignment:
    """Represents a cluster-level principal assignment."""
    # Read-only properties
    id: str  # Resource ID
    name: str  # Assignment name
    type: str  # Resource type
    provisioning_state: ProvisioningState  # Provisioning state
    principal_name: str  # Principal display name
    tenant_name: str  # Tenant display name
    aad_object_id: str  # Azure AD object ID
    
    # Configuration properties
    principal_id: str  # Principal object ID
    role: ClusterPrincipalRole  # Principal role
    tenant_id: str  # Principal tenant ID
    principal_type: PrincipalType  # Principal type

class DatabasePrincipalAssignment:
    """Represents a database-level principal assignment."""
    # Read-only properties
    id: str  # Resource ID
    name: str  # Assignment name
    type: str  # Resource type
    provisioning_state: ProvisioningState  # Provisioning state
    principal_name: str  # Principal display name
    tenant_name: str  # Tenant display name
    aad_object_id: str  # Azure AD object ID
    
    # Configuration properties
    principal_id: str  # Principal object ID
    role: DatabasePrincipalRole  # Principal role
    tenant_id: str  # Principal tenant ID
    principal_type: PrincipalType  # Principal type

class PrivateEndpointConnection:
    """Represents a private endpoint connection."""
    # Read-only properties
    id: str  # Resource ID
    name: str  # Connection name
    type: str  # Resource type
    provisioning_state: ProvisioningState  # Provisioning state
    
    # Configuration properties
    private_endpoint: PrivateEndpointProperty  # Private endpoint details
    private_link_service_connection_state: PrivateLinkServiceConnectionStateProperty  # Connection state
    group_ids: List[str]  # Group IDs

class ManagedPrivateEndpoint:
    """Represents a managed private endpoint."""
    # Read-only properties
    id: str  # Resource ID
    name: str  # Endpoint name
    type: str  # Resource type
    provisioning_state: ProvisioningState  # Provisioning state
    
    # Configuration properties
    private_link_resource_id: str  # Target resource ID
    private_link_resource_region: str  # Target resource region
    group_id: str  # Private link sub-resource group ID
    request_message: str  # Request message
    fqdns: List[str]  # Fully qualified domain names

class PrivateLinkResource:
    """Represents a private link resource."""
    # Read-only properties
    id: str  # Resource ID
    name: str  # Resource name
    type: str  # Resource type
    group_id: str  # Group ID
    required_members: List[str]  # Required members
    required_zone_names: List[str]  # Required DNS zone names

class PrivateEndpointProperty:
    """Private endpoint property details."""
    id: str  # Private endpoint resource ID

class PrivateLinkServiceConnectionStateProperty:
    """Private link service connection state."""
    status: str  # Connection status
    description: str  # Connection description
    actions_required: str  # Required actions

from enum import Enum

class ClusterPrincipalRole(str, Enum):
    """Cluster principal role values."""
    ALL_DATABASES_ADMIN = "AllDatabasesAdmin"
    ALL_DATABASES_VIEWER = "AllDatabasesViewer"

class DatabasePrincipalRole(str, Enum):
    """Database principal role values."""
    ADMIN = "Admin"
    INGESTOR = "Ingestor"
    MONITOR = "Monitor"
    USER = "User"
    UNRESTRICTED_VIEWER = "UnrestrictedViewer"
    VIEWER = "Viewer"

class PrincipalType(str, Enum):
    """Principal type values."""
    APP = "App"
    GROUP = "Group"
    USER = "User"

Install with Tessl CLI