import functools
import sys
import threading
from contextlib import contextmanager
from .base import DEBUG, Logger
class _SlowContextNotifier:
def __init__(self, threshold, func):
self.timer = threading.Timer(threshold, func)
def __enter__(self):
self.timer.start()
return self
def __exit__(self, *_):
self.timer.cancel()
_slow_logger = Logger("Slow")
[docs]
def logged_if_slow(*args, **kwargs):
"""Context manager that logs if operations within take longer than
`threshold` seconds.
:param threshold: Number of seconds (or fractions thereof) allwoed before
logging occurs. The default is 1 second.
:param logger: :class:`~logbook.Logger` to use. The default is a 'slow'
logger.
:param level: Log level. The default is `DEBUG`.
:param func: (Deprecated). Function to call to perform logging.
The remaining parameters are passed to the
:meth:`~logbook.base.LoggerMixin.log` method.
"""
threshold = kwargs.pop("threshold", 1)
func = kwargs.pop("func", None)
if func is None:
logger = kwargs.pop("logger", _slow_logger)
level = kwargs.pop("level", DEBUG)
func = functools.partial(logger.log, level, *args, **kwargs)
else:
if "logger" in kwargs or "level" in kwargs:
raise TypeError(
"If using deprecated func parameter, 'logger' and"
" 'level' arguments cannot be passed."
)
func = functools.partial(func, *args, **kwargs)
return _SlowContextNotifier(threshold, func)
class _Local(threading.local):
enabled = True
_local = _Local()
[docs]
@contextmanager
def suppressed_deprecations():
"""Disables deprecation messages temporarily
>>> with suppressed_deprecations():
... call_some_deprecated_logic()
.. versionadded:: 0.12
"""
prev_enabled = _local.enabled
_local.enabled = False
try:
yield
finally:
_local.enabled = prev_enabled
_deprecation_logger = Logger("deprecation")
_deprecation_locations = set()
def forget_deprecation_locations():
_deprecation_locations.clear()
def _write_deprecations_if_needed(message, frame_correction):
if not _local.enabled:
return
caller_location = _get_caller_location(frame_correction=frame_correction + 1)
if caller_location not in _deprecation_locations:
_deprecation_logger.warning(message, frame_correction=frame_correction + 1)
_deprecation_locations.add(caller_location)
def log_deprecation_message(message, frame_correction=0):
_write_deprecations_if_needed(
f"Deprecation message: {message}", frame_correction=frame_correction + 1
)
class _DeprecatedFunction:
def __init__(self, func, message, obj=None, objtype=None):
super().__init__()
self._func = func
self._message = message
self._obj = obj
self._objtype = objtype
def _get_underlying_func(self):
returned = self._func
if isinstance(returned, classmethod):
if hasattr(returned, "__func__"):
returned = returned.__func__
else:
returned = returned.__get__(self._objtype).__func__
return returned
def __call__(self, *args, **kwargs):
func = self._get_underlying_func()
warning = f"{self._get_func_str()} is deprecated."
if self._message is not None:
warning += f" {self._message}"
_write_deprecations_if_needed(warning, frame_correction=+1)
if self._obj is not None:
return func(self._obj, *args, **kwargs)
elif self._objtype is not None:
return func(self._objtype, *args, **kwargs)
return func(*args, **kwargs)
def _get_func_str(self):
func = self._get_underlying_func()
if self._objtype is not None:
return f"{self._objtype.__name__}.{func.__name__}"
return f"{func.__module__}.{func.__name__}"
def __get__(self, obj, objtype):
return self.bound_to(obj, objtype)
def bound_to(self, obj, objtype):
return _DeprecatedFunction(self._func, self._message, obj=obj, objtype=objtype)
@property
def __name__(self):
return self._get_underlying_func().__name__
@property
def __doc__(self):
if returned := self._get_underlying_func().__doc__: # pylint: disable=no-member
returned += "\n.. deprecated\n" # pylint: disable=no-member
if self._message:
returned += f" {self._message}" # pylint: disable=no-member
return returned
@__doc__.setter
def __doc__(self, doc):
self._get_underlying_func().__doc__ = doc
[docs]
def deprecated(func=None, message=None):
"""Marks the specified function as deprecated, and emits a warning when
it's called.
>>> @deprecated(message='No longer supported')
... def deprecated_func():
... pass
This will cause a warning log to be emitted when the function gets called,
with the correct filename/lineno.
.. versionadded:: 0.12
"""
if isinstance(func, str):
assert message is None
message = func
func = None
if func is None:
return functools.partial(deprecated, message=message)
return _DeprecatedFunction(func, message)
def _get_caller_location(frame_correction):
frame = sys._getframe(frame_correction + 1) # pylint: disable=protected-access
try:
return (frame.f_code.co_name, frame.f_lineno)
finally:
del frame