Message339201
I think Caleb's "sample_before_and_after" idea hints that this may be an idea that could benefit from the ExitStack treatment where contextlib provides a building block that handles the interaction with the context management machinery, with the documentation shows recipes for particular use cases (such as implementing timers for blocks of code)
For example:
class ContextMonitor:
"""Invoke given sampling callbacks when a context is entered and exited, and optionally combine the sampling results
*on_entry*: zero-arg sampling function to call on context entry
*on_exit*: zero-arg sampling function to call on context exit
*combine_samples*: two-arg sample combination function. If not given, samples are combined as 2-tuples.
*keep_results*: whether to keep results for retrieval via ``pop_combined_result()``
Instances of this context manager are reusable and reentrant.
"""
def __init__(self, on_entry=None, on_exit=None, combine_samples=None, *, keep_results = False):
self._on_entry = lambda:None if on_entry is None else on_entry
self._on_exit = lambda:None if on_exit is None else on_exit
self._combine_samples = lambda *args: args if combine_samples is None else combine_samples
self._entry_samples = []
self._keep_results = keep_results
self._combined_results = [] if keep_results else None
@classmethod
def sample(cls, on_event=None, check_results=None):
"""Context monitor that uses the same sampling callback on entry and exit"""
return cls(on_event, on_event, check_results)
def pop_combined_result(self):
"""Pops the last combined result. Raises RuntimeError if no results are available"""
results = self._combined_results
if not results:
raise RuntimeError("No sample results to report")
return self.checked_results.pop()
def __enter__(self):
self._entry_samples.append(self._on_entry())
return self
def __exit__(self, *args):
entry_sample = self._entry_samples.pop()
exit_sample = self._on_exit()
result = self._combine_samples(entry_sample, exit_sample)
if self._keep_results:
self._combined_results.append(result)
And then a recipe like the following (adapted from Caleb's example):
def log_if_slow(logger_name, msg, *args, threshold_sec=1.0, **kwargs):
""""Context manager that logs monitored blocks that take too long"""
logger = logging.getLogger(logger_name)
if not logger.isEnabledFor(logging.INFO):
# Avoid the timer overhead if the logger will never print anything
return nullcontext()
def _log_slow_blocks(start, end):
duration = end - start
if dt >= threshold_sec:
logger.info(msg, duration, *args, **kwargs)
return ContextMonitor.sample(time.perfcounter, _log_slow_blocks)
with log_if_slow(__name__, 'Took longer to run than expected: %.4g s'):
...
The question is whether anyone would actually find it easier to learn to use an API like ContextMonitor over just writing their own generator based context manager.
Depending on how familiar they are with the context management protocol, it's plausible that they would, as the construction API only asks a few questions:
* whether to use the same sampling function on entry and exit or different ones
* which sampling function(s) to use
* how to combine the two samples into a single result (defaulting to producing a two-tuple)
* whether to keep the results around for later retrieval (useful if you want to post-process the samples rather than dealing with them directly in the sample combination callback) |
|
Date |
User |
Action |
Args |
2019-03-30 14:11:14 | ncoghlan | set | recipients:
+ ncoghlan, tim.peters, georg.brandl, rhettinger, giampaolo.rodola, rbcollins, ezio.melotti, steven.daprano, r.david.murray, THRlWiTi, cjrh, martin.panter, dmoore, cheryl.sabella, xgdomingo |
2019-03-30 14:11:14 | ncoghlan | set | messageid: <1553955074.68.0.658077210364.issue19495@roundup.psfhosted.org> |
2019-03-30 14:11:14 | ncoghlan | link | issue19495 messages |
2019-03-30 14:11:14 | ncoghlan | create | |
|