Telemetry

rait_connector.telemetry

Azure Monitor telemetry client for fetching application insights data.

TelemetryClient

Client for fetching Azure Monitor telemetry data.

This client wraps Azure's LogsQueryClient to fetch telemetry from Application Insights tables like AppDependencies, AppExceptions, and AppAvailabilityResults.

Example

from azure.identity import DefaultAzureCredential client = TelemetryClient( ... credential=DefaultAzureCredential(), ... workspace_id="a1a4fc6d-..." ... ) data = client.fetch_all( ... tables=["AppDependencies", "AppExceptions"], ... timespan=timedelta(days=1) ... )

__init__(credential, workspace_id)

Initialize telemetry client.

Parameters:

Name	Type	Description	Default
credential	TokenCredential	Azure credential for authentication	required
workspace_id	str	Azure Log Analytics workspace ID	required

fetch(table, timespan=timedelta(days=1), query=None, limit=None)

Fetch telemetry from a single table.

Parameters:

Name	Type	Description	Default
table	str	Table name (AppDependencies, AppExceptions, AppAvailabilityResults)	required
timespan	timedelta	Time range to query	timedelta(days=1)
query	Optional[str]	Custom KQL query. If None, fetches all rows (or up to limit if set)	None
limit	Optional[int]	Maximum rows to return. Defaults to None (all rows).	None

Returns:

Type	Description
List[Dict[str, Any]]	List of row dictionaries

Raises:

Type	Description
TelemetryError	If table is not supported or query fails

fetch_all(tables=None, timespan=timedelta(days=1), custom_queries=None, limit=None)

Fetch telemetry from multiple tables.

Parameters:

Name	Type	Description	Default
tables	Optional[List[str]]	List of table names. Defaults to all supported tables.	None
timespan	timedelta	Time range to query	timedelta(days=1)
custom_queries	Optional[Dict[str, str]]	Optional dict mapping table names to custom KQL queries	None
limit	Optional[int]	Maximum rows per table. Defaults to None (all rows).	None

Returns:

Type	Description
Dict[str, List[Dict[str, Any]]]	Dictionary mapping table names to lists of row dictionaries

Example

data = client.fetch_all( ... tables=["AppDependencies", "AppExceptions"], ... timespan=timedelta(days=2) ... ) print(data.keys()) dict_keys(['AppDependencies', 'AppExceptions'])