|
""" |
|
Sample utils module (not expected to work) |
|
""" |
|
|
|
import functools |
|
import logging |
|
import traceback |
|
|
|
from laboratory import Experiment |
|
|
|
_log = logging.getLogger(__name__) |
|
|
|
|
|
def deprecated(fn): |
|
""" |
|
Incomplete example of a deprecation tracker decorator. |
|
Other solutions exist in the wild, including a built-in from Python 3.13 onwards. |
|
|
|
This one in particular uses a `log_with_traceback()` function |
|
that is not shared that logs the emitted warning including |
|
up to 10 lines of the traceback to aid in tracking down callers. |
|
|
|
- https://pypi.org/project/Deprecated/ |
|
- https://docs.python.org/3.13/library/warnings.html#warnings.deprecated |
|
""" |
|
@functools.wraps(fn) |
|
def wrapper(*args, **kwargs): |
|
err = DeprecationWarning(fn.__name__) |
|
exc_lines = traceback.format_exception_only(type(err), err) |
|
log_dict = { |
|
'msg': 'call_to_deprecated_function', |
|
'qualname': fn.__qualname__, |
|
'module': fn.__module__, |
|
'name': fn.__name__, |
|
} |
|
log_with_traceback( |
|
log_dict, extra_stack=exc_lines, |
|
offset=1, limit=10, |
|
) |
|
return fn(*args, **kwargs) |
|
return wrapper |
|
|
|
|
|
class Probe(Experiment): |
|
def __init__(self, enable_probing, *args, **kwargs): |
|
self.enable_probing = enable_probing |
|
self.compare_results = None |
|
super().__init__(*args, **kwargs) |
|
|
|
def enabled(self): |
|
return self.enable_probing |
|
|
|
def publish(self, result): |
|
if not result.match: |
|
_log.warning( |
|
{ |
|
'evt': "probe_result_mismatch", |
|
'msg': f"results did not match for probe: {self.name}", |
|
'probe_name': self.name, |
|
'control': { |
|
'function_name': result.control.name, |
|
'exception': result.control.exception, |
|
'context': result.control.get_context(), |
|
'value': result.control.value, |
|
}, |
|
'candidates': { |
|
i: { |
|
'function_name': candidate.name, |
|
'exception': candidate.exception, |
|
'exc_info': candidate.exc_info, |
|
'value': candidate.value, |
|
'diff': deepdiff.DeepDiff( |
|
result.control.value, |
|
candidate.value, |
|
verbose_level=0, |
|
), |
|
} for i, candidate in enumerate(result.candidates) |
|
}, |
|
} |
|
) |
|
|
|
self._emit_metric(self.name, result.control.name, result.control.duration) |
|
for candidate in result.candidates: |
|
self._emit_metric(self.name, candidate.name, candidate.duration) |
|
|
|
# Alternate implementation of `publish()` that uses console printing, and no deep-diffing. |
|
# def publish(self, result): |
|
# if not result.match: |
|
# print("Probe mismatch detected:", result) |
|
# for candidate in result.candidates: |
|
# print("candidate func\t{}\t{}".format(candidate.name, candidate.duration)) |
|
# print("control func\t{}\t{}".format(result.control.name, result.control.duration)) |