adx_insert_retrieve.py

from datetime import timedelta

import pandas
from azure.kusto.data.exceptions import KustoServiceError
from azure.kusto.data.helpers import dataframe_from_result_table
from azure.kusto.data.request import (ClientRequestProperties, KustoClient,
                                      KustoConnectionStringBuilder)
from azure.kusto.ingest import (BlobDescriptor, DataFormat, FileDescriptor,
                                IngestionMappingType, IngestionProperties,
                                KustoIngestClient, KustoStreamingIngestClient,
                                ReportLevel, StreamDescriptor)

# In case you want to authenticate with AAD application.
client_id = "5d32b1b5-881f-41ee-8826-3f3dff8557b8"
client_secret = "xxxxxxxxxxx"

# read more at https://docs.microsoft.com/en-us/onedrive/find-your-office-365-tenant-id
authority_id = "72f988bf-86f1-41af-91ab-2d7cd011db47"

clusterIngest = "https://ingest-dataexplorerpr.centralus.kusto.windows.net"

kcsbingest = KustoConnectionStringBuilder.with_aad_application_key_authentication(
    clusterIngest, client_id, client_secret, authority_id
)

ingestClient = KustoIngestClient(kcsbingest)

ingestion_props = IngestionProperties(
    database="prdatabase",
    table="TestData",
    dataFormat=DataFormat.CSV,
    # in case status update for success are also required
    # reportLevel=ReportLevel.FailuresAndSuccesses,
    # in case a mapping is required
    # ingestionMappingReference="{json_mapping_that_already_exists_on_table}"
    # ingestionMappingType=IngestionMappingType.Json
)

fields = ["Name", "Metric", "Source"]
rows = [["p1", 23, "SDK"], ["p2", 25, "SDK"]]

df = pandas.DataFrame(data=rows, columns=fields)

ingestClient.ingest_from_dataframe(df, ingestion_properties=ingestion_props)





cluster = "https://dataexplorerpr.centralus.kusto.windows.net"

kcsb = KustoConnectionStringBuilder.with_aad_application_key_authentication(
    cluster, client_id, client_secret, authority_id
)

client = KustoClient(kcsb)

# once authenticated, usage is as following
db = "prdatabase"
query = "TestData | take 10"

response = client.execute(db, query)

# we also support dataframes:
dataframe = dataframe_from_result_table(response.primary_results[0])

print(dataframe)