Skip to content

deep.processor.context.trigger_context

A context for the handling of a trigger.

TriggerContext

Context for a trigger.

A context is created in a valid location is triggered. This context is then used to process all the actions, collect the data and ship of the results.

Source code in deep/processor/context/trigger_context.py 
 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
class TriggerContext:
    """
    Context for a trigger.

    A context is created in a valid location is triggered. This context is then used to process all the actions,
    collect the data and ship of the results.
    """

    def __init__(self, config: ConfigService, push_service: PushService, frame: FrameType, event: str, arg: any):
        """
        Create a new trigger context.

        :param config: the config service
        :param push_service: the push service
        :param frame: the frame data
        :param event: the trigger event
        :param arg: the trigger arg
        """
        self.__push_service = push_service
        self.__event = event
        self.__frame = frame
        self.__arg = arg
        self.__config = config
        self.__results: List[ActionResult] = []
        self.__ts: int = time_ns()
        self.__id: str = str(uuid.uuid4())
        self.__frame_collector: Optional[FrameCollector] = None
        self.var_cache = VariableCacheProvider()
        self.callbacks: List[ActionCallback] = []
        self.vars: Dict[str: Variable] = {}

    def __enter__(self):
        """Start the 'with' statement and open this context."""
        return self

    def __exit__(self, exception_type, exception_value, exception_traceback):
        """Complete the 'with' statement, and close this context."""
        for result in self.__results:
            try:
                new_callback = result.process(self)
                if new_callback is not None:
                    self.callbacks.append(new_callback)
            except Exception:
                deep.logging.exception("failed to process result {}", result)

    @property
    def id(self):
        """The trigger context id."""
        return self.__id

    @property
    def file_name(self):
        """The trigger location source file name."""
        return self.__frame.f_code.co_filename

    @property
    def locals(self) -> Dict[str, any]:
        """The local frame variables."""
        return self.__frame.f_locals

    @property
    def ts(self):
        """The timestamp in nanoseconds for this trigger."""
        return self.__ts

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

    @property
    def frame(self):
        """The raw frame data."""
        return self.__frame

    @property
    def config(self) -> ConfigService:
        """The config service."""
        return self.__config

    @property
    def arg(self):
        """The trigger arg value."""
        return self.__arg

    @property
    def event(self):
        """The trigger event value."""
        return self.__event

    def action_context(self, action: 'LocationAction') -> 'ActionContext':
        """
        Create an action context from this context, for the provided action.

        :param action: the action
        :return: the new action context.
        """
        if action.action_type == LocationAction.ActionType.Snapshot:
            return SnapshotActionContext(self, action)
        if action.action_type == LocationAction.ActionType.Log:
            return LogActionContext(self, action)
        if action.action_type == LocationAction.ActionType.Metric:
            return MetricActionContext(self, action)
        if action.action_type == LocationAction.ActionType.Span:
            return SpanActionContext(self, action)
        return NoActionContext(self, action)

    def evaluate_expression(self, expression: str) -> any:
        """
        Evaluate an expression to a value.

        :param expression: the expression
        :return: the result of the expression, or the exception that was raised.
        """
        try:
            return eval(expression, None, self.__frame.f_locals)
        except BaseException as e:
            return e

    def attach_result(self, result: ActionResult):
        """
        Attach a result for this context.

        :param result: the new result
        """
        self.__results.append(result)

    @property
    def tracepoint_logger(self) -> TracepointLogger:
        """The tracepoint logger service."""
        return self.__config.tracepoint_logger

    @property
    def push_service(self) -> PushService:
        """The push service."""
        return self.__push_service

arg property

The trigger arg value.

config: ConfigService property

The config service.

event property

The trigger event value.

file_name property

The trigger location source file name.

frame property

The raw frame data.

id property

The trigger context id.

locals: Dict[str, any] property

The local frame variables.

push_service: PushService property

The push service.

resource property

The client resource information.

tracepoint_logger: TracepointLogger property

The tracepoint logger service.

ts property

The timestamp in nanoseconds for this trigger.

__enter__()

Start the 'with' statement and open this context.

Source code in deep/processor/context/trigger_context.py 
70
71
72
def __enter__(self):
    """Start the 'with' statement and open this context."""
    return self

__exit__(exception_type, exception_value, exception_traceback)

Complete the 'with' statement, and close this context.

Source code in deep/processor/context/trigger_context.py 
74
75
76
77
78
79
80
81
82
def __exit__(self, exception_type, exception_value, exception_traceback):
    """Complete the 'with' statement, and close this context."""
    for result in self.__results:
        try:
            new_callback = result.process(self)
            if new_callback is not None:
                self.callbacks.append(new_callback)
        except Exception:
            deep.logging.exception("failed to process result {}", result)

__init__(config, push_service, frame, event, arg)

Create a new trigger context.

:param config: the config service :param push_service: the push service :param frame: the frame data :param event: the trigger event :param arg: the trigger arg

Source code in deep/processor/context/trigger_context.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(self, config: ConfigService, push_service: PushService, frame: FrameType, event: str, arg: any):
    """
    Create a new trigger context.

    :param config: the config service
    :param push_service: the push service
    :param frame: the frame data
    :param event: the trigger event
    :param arg: the trigger arg
    """
    self.__push_service = push_service
    self.__event = event
    self.__frame = frame
    self.__arg = arg
    self.__config = config
    self.__results: List[ActionResult] = []
    self.__ts: int = time_ns()
    self.__id: str = str(uuid.uuid4())
    self.__frame_collector: Optional[FrameCollector] = None
    self.var_cache = VariableCacheProvider()
    self.callbacks: List[ActionCallback] = []
    self.vars: Dict[str: Variable] = {}

action_context(action)

Create an action context from this context, for the provided action.

:param action: the action :return: the new action context.

Source code in deep/processor/context/trigger_context.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def action_context(self, action: 'LocationAction') -> 'ActionContext':
    """
    Create an action context from this context, for the provided action.

    :param action: the action
    :return: the new action context.
    """
    if action.action_type == LocationAction.ActionType.Snapshot:
        return SnapshotActionContext(self, action)
    if action.action_type == LocationAction.ActionType.Log:
        return LogActionContext(self, action)
    if action.action_type == LocationAction.ActionType.Metric:
        return MetricActionContext(self, action)
    if action.action_type == LocationAction.ActionType.Span:
        return SpanActionContext(self, action)
    return NoActionContext(self, action)

attach_result(result)

Attach a result for this context.

:param result: the new result

Source code in deep/processor/context/trigger_context.py 
158
159
160
161
162
163
164
def attach_result(self, result: ActionResult):
    """
    Attach a result for this context.

    :param result: the new result
    """
    self.__results.append(result)

evaluate_expression(expression)

Evaluate an expression to a value.

:param expression: the expression :return: the result of the expression, or the exception that was raised.

Source code in deep/processor/context/trigger_context.py 
146
147
148
149
150
151
152
153
154
155
156
def evaluate_expression(self, expression: str) -> any:
    """
    Evaluate an expression to a value.

    :param expression: the expression
    :return: the result of the expression, or the exception that was raised.
    """
    try:
        return eval(expression, None, self.__frame.f_locals)
    except BaseException as e:
        return e