Spaces:
Build error
Build error
Validify-testbot-1
/
botbuilder-python
/libraries
/botbuilder-applicationinsights
/botbuilder
/applicationinsights
/application_insights_telemetry_client.py
# Copyright (c) Microsoft Corporation. All rights reserved. | |
# Licensed under the MIT License. | |
"""Application Insights Telemetry Client for Bots.""" | |
import traceback | |
from typing import Dict, Callable | |
from applicationinsights import TelemetryClient # pylint: disable=no-name-in-module | |
from botbuilder.core.bot_telemetry_client import ( | |
BotTelemetryClient, | |
Severity, | |
TelemetryDataPointType, | |
) | |
from .bot_telemetry_processor import BotTelemetryProcessor | |
def bot_telemetry_processor(data, context) -> bool: | |
"""Bot Telemetry Processor as a method for backward compatibility. Refer to | |
callable object :class:`BotTelemetryProcessor` for details. | |
:param data: Data from Application Insights | |
:type data: telemetry item | |
:param context: Context from Application Insights | |
:type context: context object | |
:return: determines if the event is passed to the server (False = Filtered). | |
:rtype: bool | |
""" | |
processor = BotTelemetryProcessor() | |
return processor(data, context) | |
class ApplicationInsightsTelemetryClient(BotTelemetryClient): | |
"""Application Insights Telemetry Client.""" | |
def __init__( | |
self, | |
instrumentation_key: str, | |
telemetry_client: TelemetryClient = None, | |
telemetry_processor: Callable[[object, object], bool] = None, | |
client_queue_size: int = None, | |
): | |
self._instrumentation_key = instrumentation_key | |
self._client = ( | |
telemetry_client | |
if telemetry_client is not None | |
else TelemetryClient(self._instrumentation_key) | |
) | |
if client_queue_size: | |
self._client.channel.queue.max_queue_length = client_queue_size | |
# Telemetry Processor | |
processor = ( | |
telemetry_processor | |
if telemetry_processor is not None | |
else bot_telemetry_processor | |
) | |
self._client.add_telemetry_processor(processor) | |
def track_pageview( | |
self, | |
name: str, | |
url: str, | |
duration: int = 0, | |
properties: Dict[str, object] = None, | |
measurements: Dict[str, object] = None, | |
) -> None: | |
""" | |
Send information about the page viewed in the application (a web page for instance). | |
:param name: the name of the page that was viewed. | |
:type name: str | |
:param url: the URL of the page that was viewed. | |
:type url: str | |
:param duration: the duration of the page view in milliseconds. (defaults to: 0) | |
:duration: int | |
:param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) | |
:type properties: :class:`typing.Dict[str, object]` | |
:param measurements: the set of custom measurements the client wants to attach to this data item. (defaults to: | |
None) | |
:type measurements: :class:`typing.Dict[str, object]` | |
""" | |
self._client.track_pageview(name, url, duration, properties, measurements) | |
def track_exception( | |
self, | |
exception_type: type = None, | |
value: Exception = None, | |
trace: traceback = None, | |
properties: Dict[str, object] = None, | |
measurements: Dict[str, object] = None, | |
) -> None: | |
""" | |
Send information about a single exception that occurred in the application. | |
:param exception_type: the type of the exception that was thrown. | |
:param value: the exception that the client wants to send. | |
:param trace: the traceback information as returned by :func:`sys.exc_info`. | |
:param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) | |
:type properties: :class:`typing.Dict[str, object]` | |
:param measurements: the set of custom measurements the client wants to attach to this data item. (defaults to: | |
None) | |
:type measurements: :class:`typing.Dict[str, object]` | |
""" | |
self._client.track_exception( | |
exception_type, value, trace, properties, measurements | |
) | |
def track_event( | |
self, | |
name: str, | |
properties: Dict[str, object] = None, | |
measurements: Dict[str, object] = None, | |
) -> None: | |
""" | |
Send information about a single event that has occurred in the context of the application. | |
:param name: the data to associate to this event. | |
:type name: str | |
:param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) | |
:type properties: :class:`typing.Dict[str, object]` | |
:param measurements: the set of custom measurements the client wants to attach to this data item. (defaults to: | |
None) | |
:type measurements: :class:`typing.Dict[str, object]` | |
""" | |
self._client.track_event(name, properties=properties, measurements=measurements) | |
def track_metric( | |
self, | |
name: str, | |
value: float, | |
tel_type: TelemetryDataPointType = None, | |
count: int = None, | |
min_val: float = None, | |
max_val: float = None, | |
std_dev: float = None, | |
properties: Dict[str, object] = None, | |
) -> NotImplemented: | |
""" | |
Send information about a single metric data point that was captured for the application. | |
:param name: The name of the metric that was captured. | |
:type name: str | |
:param value: The value of the metric that was captured. | |
:type value: float | |
:param tel_type: The type of the metric. (defaults to: TelemetryDataPointType.aggregation`) | |
:param count: the number of metrics that were aggregated into this data point. (defaults to: None) | |
:type count: int | |
:param min_val: the minimum of all metrics collected that were aggregated into this data point. (defaults to: | |
None) | |
:type min_val: float | |
:param max_val: the maximum of all metrics collected that were aggregated into this data point. (defaults to: | |
None) | |
:type max_val: float | |
:param std_dev: the standard deviation of all metrics collected that were aggregated into this data point. | |
(defaults to: None) | |
:type std_dev: float | |
:param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) | |
:type properties: :class:`typing.Dict[str, object]` | |
""" | |
self._client.track_metric( | |
name, value, tel_type, count, min_val, max_val, std_dev, properties | |
) | |
def track_trace( | |
self, name: str, properties: Dict[str, object] = None, severity: Severity = None | |
): | |
""" | |
Sends a single trace statement. | |
:param name: the trace statement. | |
:type name: str | |
:param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) | |
:type properties: :class:`typing.Dict[str, object]` | |
:param severity: the severity level of this trace, one of DEBUG, INFO, WARNING, ERROR, CRITICAL | |
""" | |
self._client.track_trace(name, properties, severity) | |
def track_request( | |
self, | |
name: str, | |
url: str, | |
success: bool, | |
start_time: str = None, | |
duration: int = None, | |
response_code: str = None, | |
http_method: str = None, | |
properties: Dict[str, object] = None, | |
measurements: Dict[str, object] = None, | |
request_id: str = None, | |
): | |
""" | |
Sends a single request that was captured for the application. | |
:param name: The name for this request. All requests with the same name will be grouped together. | |
:type name: str | |
:param url: The actual URL for this request (to show in individual request instances). | |
:type url: str | |
:param success: True if the request ended in success, False otherwise. | |
:type success: bool | |
:param start_time: the start time of the request. The value should look the same as the one returned by | |
:func:`datetime.isoformat`. (defaults to: None) | |
:type start_time: str | |
:param duration: the number of milliseconds that this request lasted. (defaults to: None) | |
:type duration: int | |
:param response_code: the response code that this request returned. (defaults to: None) | |
:type response_code: str | |
:param http_method: the HTTP method that triggered this request. (defaults to: None) | |
:type http_method: str | |
:param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) | |
:type properties: :class:`typing.Dict[str, object]` | |
:param measurements: the set of custom measurements the client wants to attach to this data item. (defaults to: | |
None) | |
:type measurements: :class:`typing.Dict[str, object]` | |
:param request_id: the id for this request. If None, a new uuid will be generated. (defaults to: None) | |
:type request_id: str | |
""" | |
self._client.track_request( | |
name, | |
url, | |
success, | |
start_time, | |
duration, | |
response_code, | |
http_method, | |
properties, | |
measurements, | |
request_id, | |
) | |
def track_dependency( | |
self, | |
name: str, | |
data: str, | |
type_name: str = None, | |
target: str = None, | |
duration: int = None, | |
success: bool = None, | |
result_code: str = None, | |
properties: Dict[str, object] = None, | |
measurements: Dict[str, object] = None, | |
dependency_id: str = None, | |
): | |
""" | |
Sends a single dependency telemetry that was captured for the application. | |
:param name: the name of the command initiated with this dependency call. Low cardinality value. | |
Examples are stored procedure name and URL path template. | |
:type name: str | |
:param data: the command initiated by this dependency call. Examples are SQL statement and HTTP URL with all | |
query parameters. | |
:type data: str | |
:param type_name: the dependency type name. Low cardinality value for logical grouping of dependencies and | |
interpretation of other fields like commandName and resultCode. Examples are SQL, Azure table, and HTTP. | |
(default to: None) | |
:type type_name: str | |
:param target: the target site of a dependency call. Examples are server name, host address. (default to: None) | |
:type target: str | |
:param duration: the number of milliseconds that this dependency call lasted. (defaults to: None) | |
:type duration: int | |
:param success: true if the dependency call ended in success, false otherwise. (defaults to: None) | |
:type success: bool | |
:param result_code: the result code of a dependency call. Examples are SQL error code and HTTP status code. | |
(defaults to: None) | |
:type result_code: str | |
:param properties: the set of custom properties the client wants attached to this data item. | |
(defaults to: None) | |
:type properties: :class:`typing.Dict[str, object]` | |
:param measurements: the set of custom measurements the client wants to attach to this data item. | |
(defaults to: None) | |
:type measurements: :class:`typing.Dict[str, object]` | |
:param dependency_id: the id for this dependency call. If None, a new uuid will be generated. | |
(defaults to: None) | |
:type dependency_id: str | |
""" | |
self._client.track_dependency( | |
name, | |
data, | |
type_name, | |
target, | |
duration, | |
success, | |
result_code, | |
properties, | |
measurements, | |
dependency_id, | |
) | |
def flush(self): | |
"""Flushes data in the queue. Data in the queue will be sent either immediately irrespective of what sender is | |
being used. | |
""" | |
self._client.flush() | |