Skip to content

deep.api.tracepoint.eventsnapshot

Types for the captured data.

WATCH_SOURCE_CAPTURE = 'CAPTURE' module-attribute

Watch source for captured data.

WATCH_SOURCE_LOG = 'LOG' module-attribute

Watch source for log expressions.

WATCH_SOURCE_METRIC = 'METRIC' module-attribute

Watch source for metric expressions.

WATCH_SOURCE_WATCH = 'WATCH' module-attribute

Watch source for user watch statements.

EventSnapshot

This is the model for the snapshot that is uploaded to the services.

Source code in deep/api/tracepoint/eventsnapshot.py 
 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
class EventSnapshot:
    """This is the model for the snapshot that is uploaded to the services."""

    def __init__(self, tracepoint, ts, resource, frames, var_lookup: Dict[str, 'Variable']):
        """
        Create a new snapshot object.

        :param tracepoint: the tracepoint object
        :param ts: the time in nanoseconds
        :param resource: the client resource
        :param frames: the captured frames
        :param var_lookup: the captured variables.
        """
        self._id = random.getrandbits(128)
        self._tracepoint = tracepoint
        self._var_lookup: Dict[str, 'Variable'] = var_lookup
        self._ts_nanos = ts
        self._frames = frames
        self._watches = []
        self._attributes = BoundedAttributes(immutable=False)
        self._duration_nanos = 0
        self._resource = Resource.get_empty().merge(resource)
        self._log = None

    def complete(self):
        """Close and complete the snapshot."""
        self._duration_nanos = time_ns() - self._ts_nanos

    def add_watch_result(self, watch_result: 'WatchResult'):
        """
        Append a watch result to the snapshot.

        :param watch_result: the result to append.
        :return:
        """
        self.watches.append(watch_result)

    def merge_var_lookup(self, lookup: Dict[str, 'Variable']):
        """
        Merge additional variables into the var lookup.

        :param lookup:  the values to merge
        """
        self._var_lookup.update(lookup)

    @property
    def id(self):
        """The id of this snapshot."""
        return self._id

    @property
    def id_str(self):
        """The id of this snapshot."""
        return format(self._id, "032x")

    @property
    def tracepoint(self):
        """The tracepoint that triggered this snapshot."""
        return self._tracepoint

    @property
    def var_lookup(self):
        """The captured var lookup."""
        return self._var_lookup

    @property
    def ts_nanos(self):
        """The time in nanoseconds, this snapshot was triggered."""
        return self._ts_nanos

    @property
    def frames(self):
        """The captured frames."""
        return self._frames

    @property
    def watches(self):
        """The watch results."""
        return self._watches

    @property
    def attributes(self) -> BoundedAttributes:
        """The snapshot attributes."""
        return self._attributes

    @property
    def duration_nanos(self):
        """The duration in nanoseconds."""
        return self._duration_nanos

    @property
    def resource(self):
        """The client resource information."""
        return self._resource

    @property
    def log_msg(self):
        """Get the processed log message."""
        return self._log

    @log_msg.setter
    def log_msg(self, msg):
        """Set the processed log message."""
        self._log = msg

    def __str__(self) -> str:
        """Represent this as a string."""
        return str(self.__dict__)

    def __repr__(self) -> str:
        """Represent this as a string."""
        return self.__str__()

attributes: BoundedAttributes property

The snapshot attributes.

duration_nanos property

The duration in nanoseconds.

frames property

The captured frames.

id property

The id of this snapshot.

id_str property

The id of this snapshot.

log_msg property writable

Get the processed log message.

resource property

The client resource information.

tracepoint property

The tracepoint that triggered this snapshot.

ts_nanos property

The time in nanoseconds, this snapshot was triggered.

var_lookup property

The captured var lookup.

watches property

The watch results.

__init__(tracepoint, ts, resource, frames, var_lookup)

Create a new snapshot object.

:param tracepoint: the tracepoint object :param ts: the time in nanoseconds :param resource: the client resource :param frames: the captured frames :param var_lookup: the captured variables.

Source code in deep/api/tracepoint/eventsnapshot.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(self, tracepoint, ts, resource, frames, var_lookup: Dict[str, 'Variable']):
    """
    Create a new snapshot object.

    :param tracepoint: the tracepoint object
    :param ts: the time in nanoseconds
    :param resource: the client resource
    :param frames: the captured frames
    :param var_lookup: the captured variables.
    """
    self._id = random.getrandbits(128)
    self._tracepoint = tracepoint
    self._var_lookup: Dict[str, 'Variable'] = var_lookup
    self._ts_nanos = ts
    self._frames = frames
    self._watches = []
    self._attributes = BoundedAttributes(immutable=False)
    self._duration_nanos = 0
    self._resource = Resource.get_empty().merge(resource)
    self._log = None

__repr__()

Represent this as a string.

Source code in deep/api/tracepoint/eventsnapshot.py 
135
136
137
def __repr__(self) -> str:
    """Represent this as a string."""
    return self.__str__()

__str__()

Represent this as a string.

Source code in deep/api/tracepoint/eventsnapshot.py 
131
132
133
def __str__(self) -> str:
    """Represent this as a string."""
    return str(self.__dict__)

add_watch_result(watch_result)

Append a watch result to the snapshot.

:param watch_result: the result to append. :return:

Source code in deep/api/tracepoint/eventsnapshot.py 
54
55
56
57
58
59
60
61
def add_watch_result(self, watch_result: 'WatchResult'):
    """
    Append a watch result to the snapshot.

    :param watch_result: the result to append.
    :return:
    """
    self.watches.append(watch_result)

complete()

Close and complete the snapshot.

Source code in deep/api/tracepoint/eventsnapshot.py 
50
51
52
def complete(self):
    """Close and complete the snapshot."""
    self._duration_nanos = time_ns() - self._ts_nanos

merge_var_lookup(lookup)

Merge additional variables into the var lookup.

:param lookup: the values to merge

Source code in deep/api/tracepoint/eventsnapshot.py 
63
64
65
66
67
68
69
def merge_var_lookup(self, lookup: Dict[str, 'Variable']):
    """
    Merge additional variables into the var lookup.

    :param lookup:  the values to merge
    """
    self._var_lookup.update(lookup)

StackFrame

This represents a frame of code that is being executed.

Source code in deep/api/tracepoint/eventsnapshot.py 
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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
class StackFrame:
    """This represents a frame of code that is being executed."""

    def __init__(self,
                 file_name,
                 short_path,
                 method_name,
                 line_number,
                 variables,
                 class_name,
                 is_async=False,
                 column_number=0,
                 transpiled_file_name=None,
                 transpiled_line_number=0,
                 transpiled_column_number=0,
                 app_frame=False
                 ):
        """
        Create a new StackFrame object.

        :param file_name: The full file path
        :param short_path: The short file path
        :param method_name: The method name
        :param line_number: The line number
        :param variables: Variables captured on this frame
        :param class_name: The class name
        :param is_async: Is the frame an async frame
        :param column_number: The column number
        :param transpiled_file_name: The transpiled file name
        :param transpiled_line_number: The transpiled line number
        :param transpiled_column_number: The transpiled column number
        :param app_frame: Is this frame in the user app
        """
        self._file_name = file_name
        self._short_path = short_path
        self._method_name = method_name
        self._line_number = line_number
        self._class_name = class_name
        self._async = is_async
        self._column_number = column_number
        self._transpiled_file_name = transpiled_file_name
        self._transpiled_line_number = transpiled_line_number
        self._transpiled_column_number = transpiled_column_number
        self._variables = variables
        self._app_frame = app_frame

    @property
    def file_name(self):
        """The full file path."""
        return self._file_name

    @property
    def short_path(self):
        """The short file path."""
        return self._short_path

    @property
    def method_name(self):
        """The method name."""
        return self._method_name

    @property
    def line_number(self):
        """The line number."""
        return self._line_number

    @property
    def class_name(self):
        """The class name."""
        return self._class_name

    @property
    def is_async(self):
        """Is the frame an async frame."""
        return self._async

    @property
    def column_number(self):
        """The column number."""
        return self._column_number

    @property
    def transpiled_file_name(self):
        """The transpiled file name."""
        return self._transpiled_file_name

    @property
    def transpiled_line_number(self):
        """The transpiled line number."""
        return self._transpiled_line_number

    @property
    def transpiled_column_number(self):
        """The transpiled column number."""
        return self._transpiled_column_number

    @property
    def variables(self):
        """Variables captured on this frame."""
        return self._variables

    @property
    def app_frame(self):
        """Is this frame in the user app."""
        return self._app_frame

    def __str__(self) -> str:
        """Represent this as a string."""
        return str(self.__dict__)

    def __repr__(self) -> str:
        """Represent this as a string."""
        return self.__str__()

app_frame property

Is this frame in the user app.

class_name property

The class name.

column_number property

The column number.

file_name property

The full file path.

is_async property

Is the frame an async frame.

line_number property

The line number.

method_name property

The method name.

short_path property

The short file path.

transpiled_column_number property

The transpiled column number.

transpiled_file_name property

The transpiled file name.

transpiled_line_number property

The transpiled line number.

variables property

Variables captured on this frame.

__init__(file_name, short_path, method_name, line_number, variables, class_name, is_async=False, column_number=0, transpiled_file_name=None, transpiled_line_number=0, transpiled_column_number=0, app_frame=False)

Create a new StackFrame object.

:param file_name: The full file path :param short_path: The short file path :param method_name: The method name :param line_number: The line number :param variables: Variables captured on this frame :param class_name: The class name :param is_async: Is the frame an async frame :param column_number: The column number :param transpiled_file_name: The transpiled file name :param transpiled_line_number: The transpiled line number :param transpiled_column_number: The transpiled column number :param app_frame: Is this frame in the user app

Source code in deep/api/tracepoint/eventsnapshot.py 
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
def __init__(self,
             file_name,
             short_path,
             method_name,
             line_number,
             variables,
             class_name,
             is_async=False,
             column_number=0,
             transpiled_file_name=None,
             transpiled_line_number=0,
             transpiled_column_number=0,
             app_frame=False
             ):
    """
    Create a new StackFrame object.

    :param file_name: The full file path
    :param short_path: The short file path
    :param method_name: The method name
    :param line_number: The line number
    :param variables: Variables captured on this frame
    :param class_name: The class name
    :param is_async: Is the frame an async frame
    :param column_number: The column number
    :param transpiled_file_name: The transpiled file name
    :param transpiled_line_number: The transpiled line number
    :param transpiled_column_number: The transpiled column number
    :param app_frame: Is this frame in the user app
    """
    self._file_name = file_name
    self._short_path = short_path
    self._method_name = method_name
    self._line_number = line_number
    self._class_name = class_name
    self._async = is_async
    self._column_number = column_number
    self._transpiled_file_name = transpiled_file_name
    self._transpiled_line_number = transpiled_line_number
    self._transpiled_column_number = transpiled_column_number
    self._variables = variables
    self._app_frame = app_frame

__repr__()

Represent this as a string.

Source code in deep/api/tracepoint/eventsnapshot.py 
250
251
252
def __repr__(self) -> str:
    """Represent this as a string."""
    return self.__str__()

__str__()

Represent this as a string.

Source code in deep/api/tracepoint/eventsnapshot.py 
246
247
248
def __str__(self) -> str:
    """Represent this as a string."""
    return str(self.__dict__)

Variable

This represents a captured variable value.

Source code in deep/api/tracepoint/eventsnapshot.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
class Variable:
    """This represents a captured variable value."""

    def __init__(self,
                 var_type,
                 value,
                 var_hash,
                 children,
                 truncated,
                 ):
        """
        Create a new Variable object.

        :param var_type: the type of the variable.
        :param value: the value as a string
        :param var_hash: the identity hash of the value
        :param children: list of child VariableIds
        :param truncated: is the value string truncated.
        """
        self._type = var_type
        self._value = value
        self._hash = var_hash
        self._children = children
        self._truncated = truncated

    @property
    def type(self):
        """The type of this value."""
        return self._type

    @property
    def value(self):
        """The string value of variable."""
        return self._value

    @property
    def hash(self):
        """The identity hash of this value."""
        return self._hash

    @property
    def children(self) -> List['VariableId']:
        """The children of this value."""
        return self._children

    @property
    def truncated(self):
        """Is the string value truncated."""
        return self._truncated

    def __str__(self) -> str:
        """Represent this as a string."""
        return str(self.__dict__)

    def __repr__(self) -> str:
        """Represent this as a string."""
        return self.__str__()

children: List[VariableId] property

The children of this value.

hash property

The identity hash of this value.

truncated property

Is the string value truncated.

type property

The type of this value.

value property

The string value of variable.

__init__(var_type, value, var_hash, children, truncated)

Create a new Variable object.

:param var_type: the type of the variable. :param value: the value as a string :param var_hash: the identity hash of the value :param children: list of child VariableIds :param truncated: is the value string truncated.

Source code in deep/api/tracepoint/eventsnapshot.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def __init__(self,
             var_type,
             value,
             var_hash,
             children,
             truncated,
             ):
    """
    Create a new Variable object.

    :param var_type: the type of the variable.
    :param value: the value as a string
    :param var_hash: the identity hash of the value
    :param children: list of child VariableIds
    :param truncated: is the value string truncated.
    """
    self._type = var_type
    self._value = value
    self._hash = var_hash
    self._children = children
    self._truncated = truncated

__repr__()

Represent this as a string.

Source code in deep/api/tracepoint/eventsnapshot.py 
309
310
311
def __repr__(self) -> str:
    """Represent this as a string."""
    return self.__str__()

__str__()

Represent this as a string.

Source code in deep/api/tracepoint/eventsnapshot.py 
305
306
307
def __str__(self) -> str:
    """Represent this as a string."""
    return str(self.__dict__)

VariableId

This represents a variable id, that is used for de duplication.

A VariableID is a pointer to a reference within the var lookup of the snapshot. Each VariableID can have different names and modifiers, but point to the same value.

e.g. val = "Ben" name = val

Both 'val' and 'name' have the value 'Ben' to prevent duplication of this in the var lookup, we use the VariableId to point to the value using the vid property.

Source code in deep/api/tracepoint/eventsnapshot.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
class VariableId:
    """
    This represents a variable id, that is used for de duplication.

    A VariableID is a pointer to a reference within the var lookup of the snapshot. Each VariableID can have
    different names and modifiers, but point to the same value.

    e.g.
     val = "Ben"
     name = val

    Both 'val' and 'name' have the value 'Ben' to prevent duplication of this in the var lookup, we use the
    VariableId to point to the value using the vid property.
    """

    def __init__(self,
                 vid,
                 name,
                 modifiers=None,
                 original_name=None
                 ):
        """
        Create a new variable object.

        :param vid: the variable id
        :param name: the variable name
        :param modifiers: the variable modifiers
        :param original_name: the original name
        """
        if modifiers is None:
            modifiers = []
        self._vid = vid
        self._name = name
        self._original_name = original_name
        self._modifiers = modifiers

    @property
    def vid(self):
        """Get variable id."""
        return self._vid

    @property
    def name(self):
        """Get variable name."""
        return self._name

    @property
    def original_name(self):
        """Get variable original name."""
        return self._original_name

    @property
    def modifiers(self):
        """Get variable modifiers."""
        return self._modifiers

    def __str__(self) -> str:
        """Represent this as a string."""
        return str(self.__dict__)

    def __repr__(self) -> str:
        """Represent this as a string."""
        return self.__str__()

    def __eq__(self, o) -> bool:
        """Check if the variable id matches."""
        if not isinstance(o, VariableId):
            return False

        if id(o) == id(self):
            return True

        return o._vid == self._vid and o._name == self._name and o._modifiers == self._modifiers

modifiers property

Get variable modifiers.

name property

Get variable name.

original_name property

Get variable original name.

vid property

Get variable id.

__eq__(o)

Check if the variable id matches.

Source code in deep/api/tracepoint/eventsnapshot.py 
378
379
380
381
382
383
384
385
386
def __eq__(self, o) -> bool:
    """Check if the variable id matches."""
    if not isinstance(o, VariableId):
        return False

    if id(o) == id(self):
        return True

    return o._vid == self._vid and o._name == self._name and o._modifiers == self._modifiers

__init__(vid, name, modifiers=None, original_name=None)

Create a new variable object.

:param vid: the variable id :param name: the variable name :param modifiers: the variable modifiers :param original_name: the original name

Source code in deep/api/tracepoint/eventsnapshot.py 
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
def __init__(self,
             vid,
             name,
             modifiers=None,
             original_name=None
             ):
    """
    Create a new variable object.

    :param vid: the variable id
    :param name: the variable name
    :param modifiers: the variable modifiers
    :param original_name: the original name
    """
    if modifiers is None:
        modifiers = []
    self._vid = vid
    self._name = name
    self._original_name = original_name
    self._modifiers = modifiers

__repr__()

Represent this as a string.

Source code in deep/api/tracepoint/eventsnapshot.py 
374
375
376
def __repr__(self) -> str:
    """Represent this as a string."""
    return self.__str__()

__str__()

Represent this as a string.

Source code in deep/api/tracepoint/eventsnapshot.py 
370
371
372
def __str__(self) -> str:
    """Represent this as a string."""
    return str(self.__dict__)

WatchResult

This is the result of a watch expression.

Source code in deep/api/tracepoint/eventsnapshot.py 
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
class WatchResult:
    """This is the result of a watch expression."""

    def __init__(self,
                 source: str,
                 expression: str,
                 result: Optional['VariableId'],
                 error: Optional[str] = None,
                 ):
        """
        Create new watch result.

        :param source: the watch source
        :param expression: the expression used
        :param result: the result of the expression
        :param error: the error captured during execution
        """
        self._expression = expression
        self._result = result
        self._error = error
        self.__source = source

    @property
    def expression(self) -> str:
        """The watch expression."""
        return self._expression

    @property
    def result(self) -> Optional['VariableId']:
        """The good result."""
        return self._result

    @property
    def error(self) -> Optional[str]:
        """The error."""
        return self._error

    @property
    def source(self):
        """The watch source."""
        return self.__source

error: Optional[str] property

The error.

expression: str property

The watch expression.

result: Optional[VariableId] property

The good result.

source property

The watch source.

__init__(source, expression, result, error=None)

Create new watch result.

:param source: the watch source :param expression: the expression used :param result: the result of the expression :param error: the error captured during execution

Source code in deep/api/tracepoint/eventsnapshot.py 
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
def __init__(self,
             source: str,
             expression: str,
             result: Optional['VariableId'],
             error: Optional[str] = None,
             ):
    """
    Create new watch result.

    :param source: the watch source
    :param expression: the expression used
    :param result: the result of the expression
    :param error: the error captured during execution
    """
    self._expression = expression
    self._result = result
    self._error = error
    self.__source = source