ray-project--ray/python/ray/_common/deprecation.py

import inspect
import logging
from functools import wraps
from typing import Any, Callable, Optional, TypeVar, Union, cast, overload

from ray.util import log_once
from ray.util.annotations import _mark_annotated

# TypeVar for preserving function/class signatures through decorators.
# Note: These decorators also accept properties, but we use Callable for the
# common case. Properties work at runtime but won't get full type inference.
F = TypeVar("F", bound=Callable[..., Any])

logger = logging.getLogger(__name__)

# A constant to use for any configuration that should be deprecated
# (to check, whether this config has actually been assigned a proper value or
# not).
DEPRECATED_VALUE = -1


def deprecation_warning(
    old: str,
    new: Optional[str] = None,
    *,
    help: Optional[str] = None,
    error: Optional[Union[bool, type[Exception]]] = None,
    stacklevel: int = 2,
) -> None:
    """Warns (via the `logger` object) or throws a deprecation warning/error.

    Args:
        old: A description of the "thing" that is to be deprecated.
        new: A description of the new "thing" that replaces it.
        help: An optional help text to tell the user, what to
            do instead of using `old`.
        error: Whether or which exception to raise. If True, raise ValueError.
            If False, just warn. If `error` is-a subclass of Exception,
            raise that Exception.
        stacklevel: The stacklevel to use for the warning message.
            Use 2 to point to where this function is called, 3+ to point
            further up the stack.

    Raises:
        ValueError: If `error=True`.
        Exception: Of type `error`, iff `error` is a sub-class of `Exception`.
    """
    msg = "`{}` has been deprecated.{}".format(
        old, (" Use `{}` instead.".format(new) if new else f" {help}" if help else "")
    )

    if error:
        if not isinstance(error, bool) and issubclass(error, Exception):
            # error is an Exception
            raise error(msg)
        else:
            # error is a boolean, construct ValueError ourselves
            raise ValueError(msg)
    else:
        logger.warning(
            "DeprecationWarning: " + msg + " This will raise an error in the future!",
            stacklevel=stacklevel,
        )


@overload
def Deprecated(
    old: None = None,
    *,
    new: Optional[str] = None,
    help: Optional[str] = None,
    error: Union[bool, type[Exception]],
) -> Callable[[F], F]:
    ...


@overload
def Deprecated(
    old: str,
    *,
    new: Optional[str] = None,
    help: Optional[str] = None,
    error: Union[bool, type[Exception]],
) -> Callable[[F], F]:
    ...


def Deprecated(
    old: Optional[str] = None,
    *,
    new: Optional[str] = None,
    help: Optional[str] = None,
    error: Union[bool, type[Exception]],
) -> Callable[[F], F]:
    """Decorator for documenting a deprecated class, method, or function.

    Automatically adds a `deprecation.deprecation_warning(old=...,
    error=False)` to not break existing code at this point to the decorated
    class' constructor, method, or function.

    In a next major release, this warning should then be made an error
    (by setting error=True), which means at this point that the
    class/method/function is no longer supported, but will still inform
    the user about the deprecation event.

    In a further major release, the class, method, function should be erased
    entirely from the codebase.


    .. testcode::
        :skipif: True

        from ray._common.deprecation import Deprecated
        # Deprecated class: Patches the constructor to warn if the class is
        # used.
        @Deprecated(new="NewAndMuchCoolerClass", error=False)
        class OldAndUncoolClass:
            ...

        # Deprecated class method: Patches the method to warn if called.
        class StillCoolClass:
            ...
            @Deprecated(new="StillCoolClass.new_and_much_cooler_method()",
                        error=False)
            def old_and_uncool_method(self, uncool_arg):
                ...

        # Deprecated function: Patches the function to warn if called.
        @Deprecated(new="new_and_much_cooler_function", error=False)
        def old_and_uncool_function(*uncool_args):
            ...
    """

    def _inner(obj: F) -> F:
        # A deprecated class.
        if inspect.isclass(obj):
            # Patch the class' init method to raise the warning/error.
            obj_init = obj.__init__

            def patched_init(*args, **kwargs):
                if log_once(old or obj.__name__):
                    deprecation_warning(
                        old=old or obj.__name__,
                        new=new,
                        help=help,
                        error=error,
                        stacklevel=3,
                    )
                return obj_init(*args, **kwargs)

            obj.__init__ = patched_init
            _mark_annotated(obj)
            # Return the patched class (with the warning/error when
            # instantiated).
            return obj

        # A deprecated class method or function.
        # Patch with the warning/error at the beginning.
        def _ctor(*args, **kwargs):
            if log_once(old or obj.__name__):
                deprecation_warning(
                    old=old or obj.__name__,
                    new=new,
                    help=help,
                    error=error,
                    stacklevel=3,
                )
            # Call the deprecated method/function.
            return obj(*args, **kwargs)

        # Only apply @wraps for actual callables, not properties/descriptors.
        # Setting __wrapped__ on a property causes inspect.unwrap() to return
        # the property, which breaks inspect.signature() in the tracing helper.
        if callable(obj):
            _ctor = wraps(obj)(_ctor)

        # Return the patched class method/function.
        return cast(F, _ctor)

    # Return the prepared decorator.
    return _inner