# Logbook

# Source code for logbook.utils

    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")

    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()

    @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

    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