Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
prefect / _internal / compatibility / deprecated.py
Size: Mime:
"""
Utilities for deprecated items.

When a deprecated item is used, a warning will be displayed. Warnings may not be
disabled with Prefect settings. Instead, the standard Python warnings filters can be
used.

Deprecated items require a start or end date. If a start date is given, the end date 
will be calculated 6 months later. Start and end dates are always in the format MMM YYYY
e.g. Jan 2023.
"""
import functools
import warnings
from typing import Any, Callable, Optional, Type, TypeVar

import pendulum
import pydantic

from prefect.utilities.callables import get_call_parameters
from prefect.utilities.importtools import to_qualified_name

T = TypeVar("T", bound=Callable)
M = TypeVar("M", bound=pydantic.BaseModel)


DEPRECATED_WARNING = "{name} has been deprecated{when}. It will not be available after {end_date}. {help}"
DEPRECATED_MOVED_WARNING = (
    "{name} has moved to {new_location}. It will not be available at the old import "
    "path after {end_date}. {help}"
)
DEPRECATED_DATEFMT = "MMM YYYY"  # e.g. Feb 2023


class PrefectDeprecationWarning(DeprecationWarning):
    """
    A deprecation warning.
    """


def generate_deprecation_message(
    name: str,
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
    help: str = "",
    when: str = "",
):
    if not start_date and not end_date:
        raise ValueError(
            "A start date is required if an end date is not provided. "
            f"Suggested start date is {pendulum.now().format(DEPRECATED_DATEFMT)!r}"
        )

    if not end_date:
        parsed_start_date = pendulum.from_format(start_date, DEPRECATED_DATEFMT)
        parsed_end_date = parsed_start_date.add(months=6)
        end_date = parsed_end_date.format(DEPRECATED_DATEFMT)
    else:
        # Validate format
        pendulum.from_format(end_date, DEPRECATED_DATEFMT)

    if when:
        when = " when " + when

    message = DEPRECATED_WARNING.format(
        name=name, when=when, end_date=end_date, help=help
    )
    return message.rstrip()


def deprecated_callable(
    *,
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
    stacklevel: int = 2,
    help: str = "",
) -> Callable[[T], T]:
    def decorator(fn: T):
        message = generate_deprecation_message(
            name=to_qualified_name(fn),
            start_date=start_date,
            end_date=end_date,
            help=help,
        )

        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            warnings.warn(message, PrefectDeprecationWarning, stacklevel=stacklevel)
            return fn(*args, **kwargs)

        return wrapper

    return decorator


def deprecated_parameter(
    name: str,
    *,
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
    stacklevel: int = 2,
    help: str = "",
    when: Optional[Callable[[Any], bool]] = None,
    when_message: str = "",
) -> Callable[[T], T]:
    """
    Mark a parameter in a callable as deprecated.

    Example:

        ```python

        @deprecated_parameter("y", when=lambda y: y is not None)
        def foo(x, y = None):
            return x + 1 + (y or 0)
        ```
    """

    when = when or (lambda _: True)

    def decorator(fn: T):

        message = generate_deprecation_message(
            name=f"The parameter {name!r} for {fn.__name__!r}",
            start_date=start_date,
            end_date=end_date,
            help=help,
            when=when_message,
        )

        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            try:
                parameters = get_call_parameters(fn, args, kwargs, apply_defaults=False)
            except Exception:
                # Avoid raising any parsing exceptions here
                parameters = kwargs

            if name in parameters and when(parameters[name]):
                warnings.warn(message, PrefectDeprecationWarning, stacklevel=stacklevel)

            return fn(*args, **kwargs)

        return wrapper

    return decorator


def deprecated_field(
    name: str,
    *,
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
    when_message: str = "",
    help: str = "",
    when: Optional[Callable[[Any], bool]] = None,
    stacklevel: int = 2,
):
    """
    Mark a field in a Pydantic model as deprecated.

    Raises warning only if the field is specified during init.

    Example:

        ```python

        @deprecated_field("x", when=lambda x: x is not None)
        class Model(pydantic.BaseModel)
            x: Optional[int] = None
            y: str
        ```
    """

    when = when or (lambda _: True)

    # Replaces the model's __init__ method with one that performs an additional warning check
    def decorator(model_cls: Type[M]) -> Type[M]:
        message = generate_deprecation_message(
            name=f"The field {name!r} in {model_cls.__name__!r}",
            start_date=start_date,
            end_date=end_date,
            help=help,
            when=when_message,
        )

        cls_init = model_cls.__init__

        @functools.wraps(model_cls.__init__)
        def __init__(__pydantic_self__, **data: Any) -> None:
            if name in data.keys() and when(data[name]):
                warnings.warn(message, PrefectDeprecationWarning, stacklevel=stacklevel)

            cls_init(__pydantic_self__, **data)

            field = __pydantic_self__.__fields__.get(name)
            if field is not None:
                field.field_info.extra["deprecated"] = True

        # Patch the model's init method
        model_cls.__init__ = __init__

        return model_cls

    return decorator