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

Database Management

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

Database lifecycle management within Kusto clusters including creation, configuration, principal management, and database sharing through attached database configurations. Databases are containers for tables, functions, and other data objects within a Kusto cluster.

Capabilities

Database CRUD Operations

Core database lifecycle operations for creating, reading, updating, and deleting databases within Kusto clusters.

def get(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    **kwargs
) -> Database:
    """
    Get a database in a Kusto cluster.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    
    Returns:
    Database object (ReadWriteDatabase or ReadOnlyFollowingDatabase)
    """

def begin_create_or_update(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    parameters: Database,
    caller_role: CallerRole = None,
    **kwargs
) -> LROPoller[Database]:
    """
    Create or update a database in a Kusto cluster.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - parameters: Database object with configuration
    - caller_role: Role of the caller (Admin or User)
    
    Returns:
    LROPoller for the long-running operation returning Database
    """

def begin_update(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    parameters: Database,
    caller_role: CallerRole = None,
    **kwargs
) -> LROPoller[Database]:
    """
    Update a database in a Kusto cluster.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - parameters: Database object with updates
    - caller_role: Role of the caller (Admin or User)
    
    Returns:
    LROPoller for the long-running operation returning updated Database
    """

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

Database Listing Operations

Operations to discover and list databases within Kusto clusters.

def list_by_cluster(
    resource_group_name: str,
    cluster_name: str,
    top: int = None,
    skiptoken: str = None,
    **kwargs
) -> Iterable[Database]:
    """
    List databases in a Kusto cluster.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - top: Maximum number of results to return
    - skiptoken: Token for pagination
    
    Returns:
    Iterable of Database objects
    """

Name Validation

Operations to validate database names before creation.

def check_name_availability(
    resource_group_name: str,
    cluster_name: str,
    resource_name: CheckNameRequest,
    **kwargs
) -> CheckNameResult:
    """
    Check if a database name is available in the cluster.
    
    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 and any issues
    """

Database Principal Management

Operations to manage security principals (users, groups, applications) at the database level.

def list_principals(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    **kwargs
) -> Iterable[DatabasePrincipal]:
    """
    List principals with access to the database.
    
    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 DatabasePrincipal objects
    """

def add_principals(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    database_principals_to_add: DatabasePrincipalListRequest,
    **kwargs
) -> DatabasePrincipalListResult:
    """
    Add principals to the database.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - database_principals_to_add: List of principals to add
    
    Returns:
    DatabasePrincipalListResult with operation results
    """

def remove_principals(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    database_principals_to_remove: DatabasePrincipalListRequest,
    **kwargs
) -> DatabasePrincipalListResult:
    """
    Remove principals from the database.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - database_principals_to_remove: List of principals to remove
    
    Returns:
    DatabasePrincipalListResult with operation results
    """

Database Invitation Operations

Operations to invite followers for database sharing across clusters.

def invite_follower(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    parameters: DatabaseInviteFollowerRequest,
    **kwargs
) -> DatabaseInviteFollowerResult:
    """
    Invite a follower database to replicate from this database.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - parameters: Invitation configuration
    
    Returns:
    DatabaseInviteFollowerResult with invitation details
    """

Usage Examples

Creating a Read-Write Database

from azure.mgmt.kusto.models import ReadWriteDatabase

# Configure database parameters
database_params = ReadWriteDatabase(
    location="East US",
    soft_delete_period="P365D",  # 365 days retention
    hot_cache_period="P31D"      # 31 days hot cache
)

# Create the database
poller = client.databases.begin_create_or_update(
    resource_group_name="my-resource-group",
    cluster_name="my-cluster",
    database_name="my-database",
    parameters=database_params
)

database = poller.result()
print(f"Database created: {database.name}")

Adding Database Principals

from azure.mgmt.kusto.models import (
    DatabasePrincipalListRequest,
    DatabasePrincipal,
    DatabasePrincipalRole,
    DatabasePrincipalType
)

# Create principal objects
principals_to_add = DatabasePrincipalListRequest(
    value=[
        DatabasePrincipal(
            role=DatabasePrincipalRole.ADMIN,
            name="admin-user@company.com",
            type=DatabasePrincipalType.USER,
            fqn="aaduser=admin-user@company.com",
            email="admin-user@company.com"
        ),
        DatabasePrincipal(
            role=DatabasePrincipalRole.VIEWER,
            name="analysts",
            type=DatabasePrincipalType.GROUP,
            fqn="aadgroup=analysts-group-id"
        )
    ]
)

# Add principals to database
result = client.databases.add_principals(
    resource_group_name="my-resource-group",
    cluster_name="my-cluster",
    database_name="my-database",
    database_principals_to_add=principals_to_add
)

print(f"Added {len(result.value)} principals to database")

Creating a Read-Only Following Database

from azure.mgmt.kusto.models import ReadOnlyFollowingDatabase

# Configure following database
following_db_params = ReadOnlyFollowingDatabase(
    location="West US",
    leader_cluster_resource_id="/subscriptions/sub-id/resourceGroups/rg/providers/Microsoft.Kusto/clusters/leader-cluster",
    attached_database_configuration_name="my-attached-config"
)

# Create following database
poller = client.databases.begin_create_or_update(
    resource_group_name="my-resource-group",
    cluster_name="my-follower-cluster",
    database_name="my-following-database",
    parameters=following_db_params
)

following_db = poller.result()
print(f"Following database created: {following_db.name}")

Key Types

class Database:
    """Base class for database resources."""
    # Read-only properties
    id: str  # Resource ID
    name: str  # Database name
    type: str  # Resource type
    
    # Common properties
    location: str  # Azure region
    kind: Kind  # Database kind (ReadWrite or ReadOnlyFollowing)

class ReadWriteDatabase(Database):
    """A read-write database in a Kusto cluster."""
    kind: Kind = Kind.READ_WRITE
    
    # Configuration properties
    provisioning_state: ProvisioningState  # Provisioning state
    soft_delete_period: str  # Data retention period (ISO 8601 duration)
    hot_cache_period: str  # Hot cache period (ISO 8601 duration)
    statistics: DatabaseStatistics  # Database size and statistics
    is_followed: bool  # Whether database is being followed

class ReadOnlyFollowingDatabase(Database):
    """A read-only database following a leader database."""
    kind: Kind = Kind.READ_ONLY_FOLLOWING
    
    # Following configuration
    provisioning_state: ProvisioningState  # Provisioning state
    leader_cluster_resource_id: str  # Resource ID of leader cluster
    attached_database_configuration_name: str  # Configuration name
    principals_modification_kind: PrincipalsModificationKind  # Principal handling
    table_level_sharing_properties: TableLevelSharingProperties  # Table sharing
    database_share_origin: DatabaseShareOrigin  # Share origin type
    original_database_name: str  # Original database name
    statistics: DatabaseStatistics  # Database statistics

class DatabasePrincipal:
    """Represents a security principal with database access."""
    role: DatabasePrincipalRole  # Principal role
    name: str  # Principal name
    type: DatabasePrincipalType  # Principal type
    fqn: str  # Fully qualified name
    email: str  # Email address (for users)
    app_id: str  # Application ID (for apps)
    tenant_name: str  # Tenant name

class DatabasePrincipalListRequest:
    """Request to add or remove database principals."""
    value: List[DatabasePrincipal]  # List of principals

class DatabasePrincipalListResult:
    """Result of principal add/remove operations."""
    value: List[DatabasePrincipal]  # Updated list of principals

class DatabaseInviteFollowerRequest:
    """Request to invite a follower database."""
    invitee_email: str  # Email of the invitee
    table_level_sharing_properties: TableLevelSharingProperties  # Sharing config

class DatabaseInviteFollowerResult:
    """Result of database follower invitation."""
    generated_invitation: str  # Invitation token

from enum import Enum

class Kind(str, Enum):
    """Database kind values."""
    READ_WRITE = "ReadWrite"
    READ_ONLY_FOLLOWING = "ReadOnlyFollowing"

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

class DatabasePrincipalType(str, Enum):
    """Database principal type values."""
    APP = "App"
    GROUP = "Group"
    USER = "User"

class CallerRole(str, Enum):
    """Caller role values."""
    ADMIN = "Admin"
    USER = "User"

class PrincipalsModificationKind(str, Enum):
    """Principal modification behavior."""
    UNION = "Union"
    REPLACE = "Replace"
    NONE = "None"

class DatabaseShareOrigin(str, Enum):
    """Database share origin values."""
    DIRECT = "Direct"
    DATA_SHARE = "DataShare"
    OTHER = "Other"

Install with Tessl CLI