Skip to content

deep.processor.context.log_action

Handling for log actions.

LogActionContext

Bases: ActionContext

The context for processing a log action.

Source code in deep/processor/context/log_action.py 
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
class LogActionContext(ActionContext):
    """The context for processing a log action."""

    def _process_action(self):
        log_msg = self.location_action.config.get(LOG_MSG)
        log, watches, vars_ = self.process_log(log_msg)
        self.trigger_context.attach_result(LogActionResult(self.location_action, log))

    def process_log(self, log_msg) -> Tuple[str, List['WatchResult'], Dict[str, 'Variable']]:
        """
        Process the log message.

        :param log_msg: the configure log message

        :returns:
            (str) process_log: the result of the processed log
            (list) watch: the watch results from the expressions
            (dic) vars: the collected variables
        """
        ctx_self = self
        watch_results = []
        _var_lookup = {}

        class FormatDict(dict):
            """This type is used in the log process to ensure that missing values are formatted don't error."""

            def __missing__(self, key):
                return "{%s}" % key

        import string

        class FormatExtractor(string.Formatter):
            """
            Allows logs to be formatted correctly.

            This type allows us to use watches within log strings and collect the watch
            as well as interpolate the values.
            """

            def get_field(self, field_name, args, kwargs):
                # evaluate watch
                watch, var_lookup, log_str = ctx_self.eval_watch(field_name, WATCH_SOURCE_LOG)
                # collect data
                watch_results.append(watch)
                _var_lookup.update(var_lookup)

                return log_str, field_name

        log_msg = "[deep] %s" % FormatExtractor().vformat(log_msg, (), FormatDict(self.trigger_context.locals))
        return log_msg, watch_results, _var_lookup

process_log(log_msg)

Process the log message.

:param log_msg: the configure log message

:returns: (str) process_log: the result of the processed log (list) watch: the watch results from the expressions (dic) vars: the collected variables

Source code in deep/processor/context/log_action.py 
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
def process_log(self, log_msg) -> Tuple[str, List['WatchResult'], Dict[str, 'Variable']]:
    """
    Process the log message.

    :param log_msg: the configure log message

    :returns:
        (str) process_log: the result of the processed log
        (list) watch: the watch results from the expressions
        (dic) vars: the collected variables
    """
    ctx_self = self
    watch_results = []
    _var_lookup = {}

    class FormatDict(dict):
        """This type is used in the log process to ensure that missing values are formatted don't error."""

        def __missing__(self, key):
            return "{%s}" % key

    import string

    class FormatExtractor(string.Formatter):
        """
        Allows logs to be formatted correctly.

        This type allows us to use watches within log strings and collect the watch
        as well as interpolate the values.
        """

        def get_field(self, field_name, args, kwargs):
            # evaluate watch
            watch, var_lookup, log_str = ctx_self.eval_watch(field_name, WATCH_SOURCE_LOG)
            # collect data
            watch_results.append(watch)
            _var_lookup.update(var_lookup)

            return log_str, field_name

    log_msg = "[deep] %s" % FormatExtractor().vformat(log_msg, (), FormatDict(self.trigger_context.locals))
    return log_msg, watch_results, _var_lookup

LogActionResult

Bases: ActionResult

The result of a successful log action.

Source code in deep/processor/context/log_action.py 
 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
class LogActionResult(ActionResult):
    """The result of a successful log action."""

    def __init__(self, action: 'LocationAction', log: str):
        """
        Create a new result of a log action.

        :param action: the source action
        :param log: the log result.
        """
        self.action = action
        self.log = log

    def process(self, ctx: 'TriggerContext') -> Optional[ActionCallback]:
        """
        Process this result.

        :param ctx: the triggering context

        :return: an action callback if we need to do something at the 'end', or None
        """
        tracepoint_logger = ctx.config.tracepoint_logger
        if tracepoint_logger:
            tracepoint_logger.log_tracepoint(self.log, ctx.id, self.action.id)
        return None

__init__(action, log)

Create a new result of a log action.

:param action: the source action :param log: the log result.

Source code in deep/processor/context/log_action.py 
101
102
103
104
105
106
107
108
109
def __init__(self, action: 'LocationAction', log: str):
    """
    Create a new result of a log action.

    :param action: the source action
    :param log: the log result.
    """
    self.action = action
    self.log = log

process(ctx)

Process this result.

:param ctx: the triggering context

:return: an action callback if we need to do something at the 'end', or None

Source code in deep/processor/context/log_action.py 
111
112
113
114
115
116
117
118
119
120
121
122
def process(self, ctx: 'TriggerContext') -> Optional[ActionCallback]:
    """
    Process this result.

    :param ctx: the triggering context

    :return: an action callback if we need to do something at the 'end', or None
    """
    tracepoint_logger = ctx.config.tracepoint_logger
    if tracepoint_logger:
        tracepoint_logger.log_tracepoint(self.log, ctx.id, self.action.id)
    return None