Spaces:
Build error
Build error
File size: 4,427 Bytes
0827183 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 |
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
"""Common utilities for Django middleware."""
import collections
from applicationinsights import TelemetryClient
from applicationinsights.channel import (
AsynchronousQueue,
AsynchronousSender,
NullSender,
SynchronousQueue,
TelemetryChannel,
)
from ..processor.telemetry_processor import TelemetryProcessor
from .django_telemetry_processor import DjangoTelemetryProcessor
ApplicationInsightsSettings = collections.namedtuple(
"ApplicationInsightsSettings",
[
"ikey",
"channel_settings",
"use_view_name",
"record_view_arguments",
"log_exceptions",
],
)
ApplicationInsightsChannelSettings = collections.namedtuple(
"ApplicationInsightsChannelSettings", ["send_interval", "send_time", "endpoint"]
)
def load_settings():
from django.conf import settings # pylint: disable=import-outside-toplevel
if hasattr(settings, "APPLICATION_INSIGHTS"):
config = settings.APPLICATION_INSIGHTS
elif hasattr(settings, "APPLICATIONINSIGHTS"):
config = settings.APPLICATIONINSIGHTS
else:
config = {}
if not isinstance(config, dict):
config = {}
return ApplicationInsightsSettings(
ikey=config.get("ikey"),
use_view_name=config.get("use_view_name", False),
record_view_arguments=config.get("record_view_arguments", False),
log_exceptions=config.get("log_exceptions", True),
channel_settings=ApplicationInsightsChannelSettings(
endpoint=config.get("endpoint"),
send_interval=config.get("send_interval"),
send_time=config.get("send_time"),
),
)
saved_clients = {} # pylint: disable=invalid-name
saved_channels = {} # pylint: disable=invalid-name
def get_telemetry_client_with_processor(
key: str, channel: TelemetryChannel, telemetry_processor: TelemetryProcessor = None
) -> TelemetryClient:
"""Gets a telemetry client instance with a telemetry processor.
:param key: instrumentation key
:type key: str
:param channel: Telemetry channel
:type channel: TelemetryChannel
:param telemetry_processor: use an existing telemetry processor from caller.
:type telemetry_processor: TelemetryProcessor
:return: a telemetry client with telemetry processor.
:rtype: TelemetryClient
"""
client = TelemetryClient(key, channel)
processor = (
telemetry_processor
if telemetry_processor is not None
else DjangoTelemetryProcessor()
)
client.add_telemetry_processor(processor)
return client
def create_client(aisettings=None, telemetry_processor: TelemetryProcessor = None):
global saved_clients, saved_channels # pylint: disable=invalid-name, global-statement
if aisettings is None:
aisettings = load_settings()
if aisettings in saved_clients:
return saved_clients[aisettings]
channel_settings = aisettings.channel_settings
if channel_settings in saved_channels:
channel = saved_channels[channel_settings]
else:
sender = AsynchronousSender(service_endpoint_uri=channel_settings.endpoint)
if channel_settings.send_time is not None:
sender.send_time = channel_settings.send_time
if channel_settings.send_interval is not None:
sender.send_interval = channel_settings.send_interval
queue = AsynchronousQueue(sender)
channel = TelemetryChannel(None, queue)
saved_channels[channel_settings] = channel
ikey = aisettings.ikey
if ikey is None:
return dummy_client("No ikey specified", telemetry_processor)
client = get_telemetry_client_with_processor(
aisettings.ikey, channel, telemetry_processor
)
saved_clients[aisettings] = client
return client
def dummy_client(
reason: str, telemetry_processor: TelemetryProcessor = None
): # pylint: disable=unused-argument
"""Creates a dummy channel so even if we're not logging telemetry, we can still send
along the real object to things that depend on it to exist"""
sender = NullSender()
queue = SynchronousQueue(sender)
channel = TelemetryChannel(None, queue)
client = get_telemetry_client_with_processor(
"00000000-0000-0000-0000-000000000000", channel, telemetry_processor
)
return client
|