CtrlK
BlogDocsLog inGet started
Tessl Logo

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

scripts-extensions.mddocs/

Scripts and Extensions

Management of KQL scripts for database automation and language extensions for custom query capabilities including Python, R, and other supported languages. This enables advanced analytics scenarios and automated database operations.

Capabilities

Script Management

Operations to manage KQL scripts that automate database operations and data processing tasks.

def get(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    script_name: str,
    **kwargs
) -> Script:
    """
    Get a KQL script in a database.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - script_name: Name of the script
    
    Returns:
    Script object with script configuration and content
    """

def begin_create_or_update(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    script_name: str,
    parameters: Script,
    **kwargs
) -> LROPoller[Script]:
    """
    Create or update a KQL script in a database.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - script_name: Name of the script
    - parameters: Script object with configuration and content
    
    Returns:
    LROPoller for the long-running operation returning Script
    """

def begin_update(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    script_name: str,
    parameters: Script,
    **kwargs
) -> LROPoller[Script]:
    """
    Update a KQL script in a database.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - database_name: Name of the database
    - script_name: Name of the script
    - parameters: Script object with updates
    
    Returns:
    LROPoller for the long-running operation returning updated Script
    """

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

def list_by_database(
    resource_group_name: str,
    cluster_name: str,
    database_name: str,
    **kwargs
) -> Iterable[Script]:
    """
    List KQL scripts in a 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 Script objects
    """

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

Sandbox Custom Images

Management of custom sandbox images for specialized language extension environments.

def get(
    resource_group_name: str,
    cluster_name: str,
    sandbox_custom_image_name: str,
    **kwargs
) -> SandboxCustomImage:
    """
    Get a sandbox custom image.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - sandbox_custom_image_name: Name of the sandbox custom image
    
    Returns:
    SandboxCustomImage object
    """

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

def begin_update(
    resource_group_name: str,
    cluster_name: str,
    sandbox_custom_image_name: str,
    parameters: SandboxCustomImage,
    **kwargs
) -> LROPoller[SandboxCustomImage]:
    """
    Update a sandbox custom image.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    - sandbox_custom_image_name: Name of the sandbox custom image
    - parameters: SandboxCustomImage with updates
    
    Returns:
    LROPoller for the long-running operation returning updated SandboxCustomImage
    """

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

def list_by_cluster(
    resource_group_name: str,
    cluster_name: str,
    **kwargs
) -> Iterable[SandboxCustomImage]:
    """
    List sandbox custom images for a cluster.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - cluster_name: Name of the Kusto cluster
    
    Returns:
    Iterable of SandboxCustomImage objects
    """

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

Usage Examples

Creating a Database Setup Script

from azure.mgmt.kusto.models import Script, ScriptLevel

# Create KQL script for database setup
setup_script = Script(
    script_url="https://mystorageaccount.blob.core.windows.net/scripts/setup.kql",
    script_url_sas_token="?sv=2020-08-04&ss=b&srt=sco&sp=r&se=2024-12-31T23:59:59Z&sig=...",
    force_update_tag="v1.0",
    continue_on_errors=False
)

# Create the script
poller = client.scripts.begin_create_or_update(
    resource_group_name="my-resource-group",
    cluster_name="my-cluster",
    database_name="my-database",
    script_name="database-setup",
    parameters=setup_script
)

script = poller.result()
print(f"Script created: {script.name}")

Creating a Data Processing Script

from azure.mgmt.kusto.models import Script

# Create KQL script for data processing
processing_script = Script(
    script_content=""".create table ProcessedData (
        Timestamp: datetime,
        UserId: string,
        EventType: string,
        Value: real
    )

    .create table ProcessedData ingestion json mapping "ProcessedDataMapping" @'[
        {"column":"Timestamp","path":"$.timestamp","datatype":"datetime"},
        {"column":"UserId","path":"$.userId","datatype":"string"},
        {"column":"EventType","path":"$.eventType","datatype":"string"},
        {"column":"Value","path":"$.value","datatype":"real"}
    ]'

    .create function ProcessEvents() {
        RawEvents
        | where Timestamp > ago(1d)
        | extend ProcessedValue = Value * 1.1
        | project Timestamp, UserId, EventType, ProcessedValue
    }""",
    force_update_tag="v2.0",
    continue_on_errors=True
)

# Create the script
poller = client.scripts.begin_create_or_update(
    resource_group_name="my-resource-group",
    cluster_name="my-cluster",
    database_name="my-database",
    script_name="data-processing",
    parameters=processing_script
)

script = poller.result()
print(f"Data processing script created: {script.name}")

Creating a Custom Sandbox Image

from azure.mgmt.kusto.models import SandboxCustomImage, Language

# Create custom sandbox image for Python environment
custom_image = SandboxCustomImage(
    language=Language.PYTHON,
    language_version="3.9",
    package_requirements="""\
numpy==1.21.0
pandas==1.3.0
scikit-learn==0.24.2
matplotlib==3.4.2
seaborn==0.11.1
""",
    requirements_file_content="""\
# Additional Python packages for data science
numpy>=1.21.0
pandas>=1.3.0
scikit-learn>=0.24.2
matplotlib>=3.4.2
seaborn>=0.11.1
plotly>=5.0.0
""",
    dependencies_file_content="""\
# System dependencies
libgomp1
libatlas-base-dev
"""
)

# Create the custom image
poller = client.sandbox_custom_images.begin_create_or_update(
    resource_group_name="my-resource-group",
    cluster_name="my-cluster",
    sandbox_custom_image_name="python-datascience",
    parameters=custom_image
)

image = poller.result()
print(f"Custom sandbox image created: {image.name}")

Managing Language Extensions at Cluster Level

from azure.mgmt.kusto.models import (
    LanguageExtensionsList,
    LanguageExtension,
    LanguageExtensionName,
    LanguageExtensionImageName
)

# Add Python language extension
python_extension = LanguageExtension(
    language_extension_name=LanguageExtensionName.PYTHON,
    language_extension_image_name=LanguageExtensionImageName.PYTHON3_6_5,
    language_extension_custom_image_name="python-datascience"  # Use custom image
)

extensions_to_add = LanguageExtensionsList(
    value=[python_extension]
)

# Add language extension to cluster
poller = client.clusters.begin_add_language_extensions(
    resource_group_name="my-resource-group",
    cluster_name="my-cluster",
    language_extensions_to_add=extensions_to_add
)

poller.result()
print("Python language extension added to cluster")

# List current language extensions
extensions = list(client.clusters.list_language_extensions(
    resource_group_name="my-resource-group",
    cluster_name="my-cluster"
))

for ext in extensions:
    print(f"Extension: {ext.language_extension_name} - Image: {ext.language_extension_image_name}")

Key Types

class Script:
    """Represents a KQL script resource."""
    # Read-only properties
    id: str  # Resource ID
    name: str  # Script name
    type: str  # Resource type
    provisioning_state: ProvisioningState  # Provisioning state
    
    # Script configuration
    script_url: str  # URL to script file
    script_url_sas_token: str  # SAS token for script URL access
    script_content: str  # Inline script content (alternative to URL)
    force_update_tag: str  # Tag to force script update
    continue_on_errors: bool  # Whether to continue execution on errors

class SandboxCustomImage:
    """Represents a custom sandbox image for language extensions."""
    # Read-only properties
    id: str  # Resource ID
    name: str  # Image name
    type: str  # Resource type
    provisioning_state: ProvisioningState  # Provisioning state
    
    # Image configuration
    language: Language  # Programming language
    language_version: str  # Language version
    package_requirements: str  # Package requirements file content
    requirements_file_content: str  # Requirements file content
    dependencies_file_content: str  # System dependencies content

class LanguageExtension:
    """Represents a language extension configuration."""
    language_extension_name: LanguageExtensionName  # Extension name
    language_extension_image_name: LanguageExtensionImageName  # Image name
    language_extension_custom_image_name: str  # Custom image name

class LanguageExtensionsList:
    """List of language extensions."""
    value: List[LanguageExtension]  # Language extensions

class ScriptCheckNameRequest:
    """Request to check script name availability."""
    name: str  # Name to check
    type: str  # Resource type

class SandboxCustomImagesCheckNameRequest:
    """Request to check sandbox custom image name availability."""
    name: str  # Name to check
    type: str  # Resource type

from enum import Enum

class ScriptLevel(str, Enum):
    """Script level values."""
    DATABASE = "Database"
    CLUSTER = "Cluster"

class Language(str, Enum):
    """Programming language values."""
    PYTHON = "Python"
    R = "R"

class LanguageExtensionName(str, Enum):
    """Language extension name values."""
    PYTHON = "PYTHON"
    R = "R"

class LanguageExtensionImageName(str, Enum):
    """Language extension image name values."""
    PYTHON3_6_5 = "Python3_6_5"
    PYTHON3_9_7 = "Python3_9_7"
    PYTHON3_10_8 = "Python3_10_8"
    R3_4_4 = "R3_4_4"
    R4_0_5 = "R4_0_5"
    R4_2_5 = "R4_2_5"

Script Content Examples

Table Creation Script

// Create tables for event processing
.create table Events (
    Timestamp: datetime,
    UserId: string,
    EventType: string,
    Properties: dynamic
)

.create table EventsSummary (
    Date: datetime,
    EventType: string,
    Count: long,
    UniqueUsers: long
)

// Create ingestion mapping
.create table Events ingestion json mapping "EventsMapping" @'[
    {"column":"Timestamp","path":"$.timestamp","datatype":"datetime"},
    {"column":"UserId","path":"$.userId","datatype":"string"},
    {"column":"EventType","path":"$.eventType","datatype":"string"},
    {"column":"Properties","path":"$.properties","datatype":"dynamic"}
]'

Function Creation Script

// Create function for event aggregation
.create function EventsDailySummary() {
    Events
    | where Timestamp >= startofday(now(-1d)) and Timestamp < startofday(now())
    | summarize 
        Count = count(),
        UniqueUsers = dcount(UserId)
        by EventType, Date = bin(Timestamp, 1d)
    | order by Date desc, EventType asc
}

// Create materialized view for real-time aggregation
.create materialized-view EventsHourlySummary on table Events {
    Events
    | summarize 
        Count = count(),
        UniqueUsers = dcount(UserId),
        LastEventTime = max(Timestamp)
        by EventType, Hour = bin(Timestamp, 1h)
}

Data Retention Policy Script

// Set retention policies
.alter table Events policy retention softdelete = 30d recoverability = disabled

.alter table EventsSummary policy retention softdelete = 365d

// Set ingestion batching policy
.alter table Events policy ingestionbatching @'{"MaximumBatchingTimeSpan":"00:01:00", "MaximumNumberOfItems": 1000}'

// Enable streaming ingestion
.alter table Events policy streamingingestion enable

Install with Tessl CLI

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

docs

cluster-management.md

data-connections.md

database-management.md

index.md

monitoring-operations.md

scripts-extensions.md

security-management.md

tile.json