azure-kusto-data

import os from azure.kusto.data import KustoClient, KustoConnectionStringBuilder from azure.kusto.data.exceptions import KustoServiceError # Replace with your Kusto cluster URI CLUSTER_URI = os.environ.get('KUSTO_CLUSTER_URI', 'https://<your_cluster_name>.kusto.windows.net') # Replace with your AAD application ID (client ID) CLIENT_ID = os.environ.get('KUSTO_CLIENT_ID', 'your_aad_application_id') # Replace with your AAD application key (client secret) CLIENT_SECRET = os.environ.get('KUSTO_CLIENT_SECRET', 'your_aad_application_key') # Replace with your AAD tenant ID TENANT_ID = os.environ.get('KUSTO_TENANT_ID', 'your_aad_tenant_id') DB_NAME = 'Samples' QUERY = 'StormEvents | take 5' def main(): if 'your_aad_application_id' in CLIENT_ID or 'your_aad_application_key' in CLIENT_SECRET: print("Please set KUSTO_CLUSTER_URI, KUSTO_CLIENT_ID, KUSTO_CLIENT_SECRET, and KUSTO_TENANT_ID environment variables or replace placeholders.") return # Build connection string for AAD application key authentication kcsb = KustoConnectionStringBuilder.with_aad_application_key_authentication( CLUSTER_URI, CLIENT_ID, CLIENT_SECRET, TENANT_ID ) # It is good practice to re-use the KustoClient instance, as it maintains a pool of connections. try: with KustoClient(kcsb) as client: print(f"Executing query on database '{DB_NAME}'...") response = client.execute(DB_NAME, QUERY) for row in response.primary_results[0]: print(f"Timestamp: {row['StartTime']}, EventType: {row['EventType']}, State: {row['State']}") except KustoServiceError as e: print(f"Kusto service error: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") if __name__ == '__main__': main()