Skip to content

deep.config.config_service

Service for handling deep config.

ConfigService

This is the main service that handles config for DEEP.

Source code in deep/config/config_service.py 
 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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
class ConfigService:
    """This is the main service that handles config for DEEP."""

    def __init__(self, custom: Dict[str, any] = None, tracepoints=TracepointConfigService()):
        """
        Create a new config object.

        :param custom: any custom values that are passed to DEEP
        """
        if custom is None:
            custom = {}
        self._plugins = []
        self.__custom = custom
        self._resource = None
        self._tracepoint_config = tracepoints

    def __getattribute__(self, name: str) -> Any:
        """
        Get attribute from config.

        A custom attribute processor to load the config values

        :param name: the key to load
        :return: the loaded value or None
        """
        attr = None
        try:
            # first we try to load from our selves. This handles things like getting 'self.custom' etc.
            attr = super().__getattribute__(name)
        except AttributeError:
            # if we get here then we are not an attribute on 'self'
            # so look in the custom map
            if self.__custom is not None and name in self.__custom:
                attr = self.__custom[name]

            # if not in custom then load from 'deep.config'
            if attr is None:
                from deep import config
                has_attr = hasattr(config, name)
                if not has_attr:
                    # attribute is no in 'deep.config', so look in env
                    from_env = os.getenv("DEEP_%s" % name, None)
                    if from_env is None:
                        # not found in env - log and return none
                        logging.warning("Unrecognised config key: %s", name)
                        return None
                    else:
                        # if loaded from env, then cannot be function
                        return from_env
                attr = getattr(config, name, None)

            # some can be callable - if is callable call it
            if callable(attr):
                return attr()

        return attr

    def __setattr__(self, name: str, value: Any) -> None:
        """Set attribute on this config."""
        super().__setattr__(name, value)

    def set_task_handler(self, task_handler):
        """
        Set the task handler to use.

        :param task_handler: the taskhandler
        """
        self._tracepoint_config.set_task_handler(task_handler)

    @property
    def resource(self) -> Resource:
        """Get the resource that describes this client."""
        return self._resource

    @resource.setter
    def resource(self, new_resource):
        """Set the resource that describes this client."""
        self._resource = new_resource

    @property
    def plugins(self) -> List[Plugin]:
        """Get the active deep client plugins."""
        return self._plugins

    @plugins.setter
    def plugins(self, plugins):
        """Set the active deep client plugins."""
        self._plugins = plugins

    @property
    def tracepoints(self) -> 'TracepointConfigService':
        """The tracepoint config service."""
        return self._tracepoint_config

    def add_listener(self, listener: 'ConfigUpdateListener'):
        """
        Add a new listener to the config.

        :param listener: the listener to add
        """
        self._tracepoint_config.add_listener(listener)

    @property
    def tracepoint_logger(self) -> TracepointLogger:
        """Get the tracepoint logger."""
        return self._find_plugin(TracepointLogger)

    @property
    def resource_providers(self) -> Generator[ResourceProvider, None, None]:
        """Generator for available resource providers."""
        return self.__plugin_generator(ResourceProvider)

    @property
    def snapshot_decorators(self) -> Generator[SnapshotDecorator, None, None]:
        """Generator for snapshot decorators."""
        return self.__plugin_generator(SnapshotDecorator)

    @property
    def metric_processors(self) -> Generator[MetricProcessor, None, None]:
        """Generator for snapshot decorators."""
        return self.__plugin_generator(MetricProcessor)

    @property
    def has_metric_processor(self) -> bool:
        """Is there a configured metric processor."""
        return self._find_plugin(MetricProcessor) is not None

    @property
    def span_processors(self) -> Generator[SpanProcessor, None, None]:
        """Generator for snapshot decorators."""
        return self.__plugin_generator(SpanProcessor)

    @property
    def has_span_processor(self) -> bool:
        """Is there a configured metric processor."""
        return self._find_plugin(SpanProcessor) is not None

    def is_app_frame(self, filename: str) -> Tuple[bool, Optional[str]]:
        """
        Check if the current frame is a user application frame.

        :param filename: the frame file name
        :return: True if add frame, else False
        """
        in_app_include = self.IN_APP_INCLUDE
        in_app_exclude = self.IN_APP_EXCLUDE

        for path in in_app_exclude:
            if filename.startswith(path):
                return False, path

        for path in in_app_include:
            if filename.startswith(path):
                return True, path

        if filename.startswith(self.APP_ROOT):
            return True, self.APP_ROOT

        return False, None

    def _find_plugin(self, plugin_type) -> PLUGIN_TYPE:
        return next(self.__plugin_generator(plugin_type), None)

    def _find_plugins(self, plugin_type) -> List[PLUGIN_TYPE]:
        plugins = []
        for plugin in self.__plugin_generator(plugin_type):
            plugins.append(plugin)
        return plugins

    def __plugin_generator(self, plugin_type) -> Generator[PLUGIN_TYPE, None, None]:
        for plugin in self._plugins:
            if isinstance(plugin, plugin_type):
                yield plugin

has_metric_processor: bool property

Is there a configured metric processor.

has_span_processor: bool property

Is there a configured metric processor.

metric_processors: Generator[MetricProcessor, None, None] property

Generator for snapshot decorators.

plugins: List[Plugin] property writable

Get the active deep client plugins.

resource: Resource property writable

Get the resource that describes this client.

resource_providers: Generator[ResourceProvider, None, None] property

Generator for available resource providers.

snapshot_decorators: Generator[SnapshotDecorator, None, None] property

Generator for snapshot decorators.

span_processors: Generator[SpanProcessor, None, None] property

Generator for snapshot decorators.

tracepoint_logger: TracepointLogger property

Get the tracepoint logger.

tracepoints: TracepointConfigService property

The tracepoint config service.

__getattribute__(name)

Get attribute from config.

A custom attribute processor to load the config values

:param name: the key to load :return: the loaded value or None

Source code in deep/config/config_service.py 
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
def __getattribute__(self, name: str) -> Any:
    """
    Get attribute from config.

    A custom attribute processor to load the config values

    :param name: the key to load
    :return: the loaded value or None
    """
    attr = None
    try:
        # first we try to load from our selves. This handles things like getting 'self.custom' etc.
        attr = super().__getattribute__(name)
    except AttributeError:
        # if we get here then we are not an attribute on 'self'
        # so look in the custom map
        if self.__custom is not None and name in self.__custom:
            attr = self.__custom[name]

        # if not in custom then load from 'deep.config'
        if attr is None:
            from deep import config
            has_attr = hasattr(config, name)
            if not has_attr:
                # attribute is no in 'deep.config', so look in env
                from_env = os.getenv("DEEP_%s" % name, None)
                if from_env is None:
                    # not found in env - log and return none
                    logging.warning("Unrecognised config key: %s", name)
                    return None
                else:
                    # if loaded from env, then cannot be function
                    return from_env
            attr = getattr(config, name, None)

        # some can be callable - if is callable call it
        if callable(attr):
            return attr()

    return attr

__init__(custom=None, tracepoints=TracepointConfigService())

Create a new config object.

:param custom: any custom values that are passed to DEEP

Source code in deep/config/config_service.py 
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, custom: Dict[str, any] = None, tracepoints=TracepointConfigService()):
    """
    Create a new config object.

    :param custom: any custom values that are passed to DEEP
    """
    if custom is None:
        custom = {}
    self._plugins = []
    self.__custom = custom
    self._resource = None
    self._tracepoint_config = tracepoints

__setattr__(name, value)

Set attribute on this config.

Source code in deep/config/config_service.py 
86
87
88
def __setattr__(self, name: str, value: Any) -> None:
    """Set attribute on this config."""
    super().__setattr__(name, value)

add_listener(listener)

Add a new listener to the config.

:param listener: the listener to add

Source code in deep/config/config_service.py 
123
124
125
126
127
128
129
def add_listener(self, listener: 'ConfigUpdateListener'):
    """
    Add a new listener to the config.

    :param listener: the listener to add
    """
    self._tracepoint_config.add_listener(listener)

is_app_frame(filename)

Check if the current frame is a user application frame.

:param filename: the frame file name :return: True if add frame, else False

Source code in deep/config/config_service.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def is_app_frame(self, filename: str) -> Tuple[bool, Optional[str]]:
    """
    Check if the current frame is a user application frame.

    :param filename: the frame file name
    :return: True if add frame, else False
    """
    in_app_include = self.IN_APP_INCLUDE
    in_app_exclude = self.IN_APP_EXCLUDE

    for path in in_app_exclude:
        if filename.startswith(path):
            return False, path

    for path in in_app_include:
        if filename.startswith(path):
            return True, path

    if filename.startswith(self.APP_ROOT):
        return True, self.APP_ROOT

    return False, None

set_task_handler(task_handler)

Set the task handler to use.

:param task_handler: the taskhandler

Source code in deep/config/config_service.py 
90
91
92
93
94
95
96
def set_task_handler(self, task_handler):
    """
    Set the task handler to use.

    :param task_handler: the taskhandler
    """
    self._tracepoint_config.set_task_handler(task_handler)