1# util/deprecations.py 
    2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors 
    3# <see AUTHORS file> 
    4# 
    5# This module is part of SQLAlchemy and is released under 
    6# the MIT License: https://www.opensource.org/licenses/mit-license.php 
    7# mypy: allow-untyped-defs, allow-untyped-calls 
    8 
    9"""Helpers related to deprecation of functions, methods, classes, other 
    10functionality.""" 
    11 
    12from __future__ import annotations 
    13 
    14import re 
    15from typing import Any 
    16from typing import Callable 
    17from typing import Dict 
    18from typing import Match 
    19from typing import Optional 
    20from typing import Sequence 
    21from typing import Set 
    22from typing import Tuple 
    23from typing import Type 
    24from typing import TypeVar 
    25from typing import Union 
    26 
    27from . import compat 
    28from .langhelpers import _hash_limit_string 
    29from .langhelpers import _warnings_warn 
    30from .langhelpers import decorator 
    31from .langhelpers import inject_docstring_text 
    32from .langhelpers import inject_param_text 
    33from .. import exc 
    34 
    35_T = TypeVar("_T", bound=Any) 
    36 
    37 
    38# https://mypy.readthedocs.io/en/stable/generics.html#declaring-decorators 
    39_F = TypeVar("_F", bound="Callable[..., Any]") 
    40 
    41 
    42def _warn_with_version( 
    43    msg: str, 
    44    version: str, 
    45    type_: Type[exc.SADeprecationWarning], 
    46    stacklevel: int, 
    47    code: Optional[str] = None, 
    48) -> None: 
    49    warn = type_(msg, code=code) 
    50    warn.deprecated_since = version 
    51 
    52    _warnings_warn(warn, stacklevel=stacklevel + 1) 
    53 
    54 
    55def warn_deprecated( 
    56    msg: str, version: str, stacklevel: int = 3, code: Optional[str] = None 
    57) -> None: 
    58    _warn_with_version( 
    59        msg, version, exc.SADeprecationWarning, stacklevel, code=code 
    60    ) 
    61 
    62 
    63def warn_deprecated_limited( 
    64    msg: str, 
    65    args: Sequence[Any], 
    66    version: str, 
    67    stacklevel: int = 3, 
    68    code: Optional[str] = None, 
    69) -> None: 
    70    """Issue a deprecation warning with a parameterized string, 
    71    limiting the number of registrations. 
    72 
    73    """ 
    74    if args: 
    75        msg = _hash_limit_string(msg, 10, args) 
    76    _warn_with_version( 
    77        msg, version, exc.SADeprecationWarning, stacklevel, code=code 
    78    ) 
    79 
    80 
    81def deprecated_cls( 
    82    version: str, message: str, constructor: Optional[str] = "__init__" 
    83) -> Callable[[Type[_T]], Type[_T]]: 
    84    header = ".. deprecated:: %s %s" % (version, (message or "")) 
    85 
    86    def decorate(cls: Type[_T]) -> Type[_T]: 
    87        return _decorate_cls_with_warning( 
    88            cls, 
    89            constructor, 
    90            exc.SADeprecationWarning, 
    91            message % dict(func=constructor), 
    92            version, 
    93            header, 
    94        ) 
    95 
    96    return decorate 
    97 
    98 
    99def deprecated( 
    100    version: str, 
    101    message: Optional[str] = None, 
    102    add_deprecation_to_docstring: bool = True, 
    103    warning: Optional[Type[exc.SADeprecationWarning]] = None, 
    104    enable_warnings: bool = True, 
    105) -> Callable[[_F], _F]: 
    106    """Decorates a function and issues a deprecation warning on use. 
    107 
    108    :param version: 
    109      Issue version in the warning. 
    110 
    111    :param message: 
    112      If provided, issue message in the warning.  A sensible default 
    113      is used if not provided. 
    114 
    115    :param add_deprecation_to_docstring: 
    116      Default True.  If False, the wrapped function's __doc__ is left 
    117      as-is.  If True, the 'message' is prepended to the docs if 
    118      provided, or sensible default if message is omitted. 
    119 
    120    """ 
    121 
    122    if add_deprecation_to_docstring: 
    123        header = ".. deprecated:: %s %s" % ( 
    124            version, 
    125            (message or ""), 
    126        ) 
    127    else: 
    128        header = None 
    129 
    130    if message is None: 
    131        message = "Call to deprecated function %(func)s" 
    132 
    133    if warning is None: 
    134        warning = exc.SADeprecationWarning 
    135 
    136    message += " (deprecated since: %s)" % version 
    137 
    138    def decorate(fn: _F) -> _F: 
    139        assert message is not None 
    140        assert warning is not None 
    141        return _decorate_with_warning( 
    142            fn, 
    143            warning, 
    144            message % dict(func=fn.__name__), 
    145            version, 
    146            header, 
    147            enable_warnings=enable_warnings, 
    148        ) 
    149 
    150    return decorate 
    151 
    152 
    153def moved_20( 
    154    message: str, **kw: Any 
    155) -> Callable[[Callable[..., _T]], Callable[..., _T]]: 
    156    return deprecated( 
    157        "2.0", message=message, warning=exc.MovedIn20Warning, **kw 
    158    ) 
    159 
    160 
    161def became_legacy_20( 
    162    api_name: str, alternative: Optional[str] = None, **kw: Any 
    163) -> Callable[[_F], _F]: 
    164    type_reg = re.match("^:(attr|func|meth):", api_name) 
    165    if type_reg: 
    166        type_ = {"attr": "attribute", "func": "function", "meth": "method"}[ 
    167            type_reg.group(1) 
    168        ] 
    169    else: 
    170        type_ = "construct" 
    171    message = ( 
    172        "The %s %s is considered legacy as of the " 
    173        "1.x series of SQLAlchemy and %s in 2.0." 
    174        % ( 
    175            api_name, 
    176            type_, 
    177            "becomes a legacy construct", 
    178        ) 
    179    ) 
    180 
    181    if ":attr:" in api_name: 
    182        attribute_ok = kw.pop("warn_on_attribute_access", False) 
    183        if not attribute_ok: 
    184            assert kw.get("enable_warnings") is False, ( 
    185                "attribute %s will emit a warning on read access.  " 
    186                "If you *really* want this, " 
    187                "add warn_on_attribute_access=True.  Otherwise please add " 
    188                "enable_warnings=False." % api_name 
    189            ) 
    190 
    191    if alternative: 
    192        message += " " + alternative 
    193 
    194    warning_cls = exc.LegacyAPIWarning 
    195 
    196    return deprecated("2.0", message=message, warning=warning_cls, **kw) 
    197 
    198 
    199def deprecated_params(**specs: Tuple[str, str]) -> Callable[[_F], _F]: 
    200    """Decorates a function to warn on use of certain parameters. 
    201 
    202    e.g. :: 
    203 
    204        @deprecated_params( 
    205            weak_identity_map=( 
    206                "2.0", 
    207                "the :paramref:`.Session.weak_identity_map parameter " 
    208                "is deprecated.", 
    209            ) 
    210        ) 
    211        def some_function(**kwargs): ... 
    212 
    213    """ 
    214 
    215    messages: Dict[str, str] = {} 
    216    versions: Dict[str, str] = {} 
    217    version_warnings: Dict[str, Type[exc.SADeprecationWarning]] = {} 
    218 
    219    for param, (version, message) in specs.items(): 
    220        versions[param] = version 
    221        messages[param] = _sanitize_restructured_text(message) 
    222        version_warnings[param] = exc.SADeprecationWarning 
    223 
    224    def decorate(fn: _F) -> _F: 
    225        spec = compat.inspect_getfullargspec(fn) 
    226 
    227        check_defaults: Union[Set[str], Tuple[()]] 
    228        if spec.defaults is not None: 
    229            defaults = dict( 
    230                zip( 
    231                    spec.args[(len(spec.args) - len(spec.defaults)) :], 
    232                    spec.defaults, 
    233                ) 
    234            ) 
    235            check_defaults = set(defaults).intersection(messages) 
    236            check_kw = set(messages).difference(defaults) 
    237        elif spec.kwonlydefaults is not None: 
    238            defaults = spec.kwonlydefaults 
    239            check_defaults = set(defaults).intersection(messages) 
    240            check_kw = set(messages).difference(defaults) 
    241        else: 
    242            check_defaults = () 
    243            check_kw = set(messages) 
    244 
    245        check_any_kw = spec.varkw 
    246 
    247        # latest mypy has opinions here, not sure if they implemented 
    248        # Concatenate or something 
    249        @decorator 
    250        def warned(fn: _F, *args: Any, **kwargs: Any) -> _F: 
    251            for m in check_defaults: 
    252                if (defaults[m] is None and kwargs[m] is not None) or ( 
    253                    defaults[m] is not None and kwargs[m] != defaults[m] 
    254                ): 
    255                    _warn_with_version( 
    256                        messages[m], 
    257                        versions[m], 
    258                        version_warnings[m], 
    259                        stacklevel=3, 
    260                    ) 
    261 
    262            if check_any_kw in messages and set(kwargs).difference( 
    263                check_defaults 
    264            ): 
    265                assert check_any_kw is not None 
    266                _warn_with_version( 
    267                    messages[check_any_kw], 
    268                    versions[check_any_kw], 
    269                    version_warnings[check_any_kw], 
    270                    stacklevel=3, 
    271                ) 
    272 
    273            for m in check_kw: 
    274                if m in kwargs: 
    275                    _warn_with_version( 
    276                        messages[m], 
    277                        versions[m], 
    278                        version_warnings[m], 
    279                        stacklevel=3, 
    280                    ) 
    281            return fn(*args, **kwargs)  # type: ignore[no-any-return] 
    282 
    283        doc = fn.__doc__ if fn.__doc__ is not None else "" 
    284        if doc: 
    285            doc = inject_param_text( 
    286                doc, 
    287                { 
    288                    param: ".. deprecated:: %s %s" 
    289                    % ("1.4" if version == "2.0" else version, (message or "")) 
    290                    for param, (version, message) in specs.items() 
    291                }, 
    292            ) 
    293        decorated = warned(fn) 
    294        decorated.__doc__ = doc 
    295        return decorated 
    296 
    297    return decorate 
    298 
    299 
    300def _sanitize_restructured_text(text: str) -> str: 
    301    def repl(m: Match[str]) -> str: 
    302        type_, name = m.group(1, 2) 
    303        if type_ in ("func", "meth"): 
    304            name += "()" 
    305        return name 
    306 
    307    text = re.sub(r":ref:`(.+) <.*>`", lambda m: '"%s"' % m.group(1), text) 
    308    return re.sub(r"\:(\w+)\:`~?(?:_\w+)?\.?(.+?)`", repl, text) 
    309 
    310 
    311def _decorate_cls_with_warning( 
    312    cls: Type[_T], 
    313    constructor: Optional[str], 
    314    wtype: Type[exc.SADeprecationWarning], 
    315    message: str, 
    316    version: str, 
    317    docstring_header: Optional[str] = None, 
    318) -> Type[_T]: 
    319    doc = cls.__doc__ is not None and cls.__doc__ or "" 
    320    if docstring_header is not None: 
    321        if constructor is not None: 
    322            docstring_header %= dict(func=constructor) 
    323 
    324        if issubclass(wtype, exc.Base20DeprecationWarning): 
    325            docstring_header += ( 
    326                " (Background on SQLAlchemy 2.0 at: " 
    327                ":ref:`migration_20_toplevel`)" 
    328            ) 
    329        doc = inject_docstring_text(doc, docstring_header, 1) 
    330 
    331        constructor_fn = None 
    332        if type(cls) is type: 
    333            clsdict = dict(cls.__dict__) 
    334            clsdict["__doc__"] = doc 
    335            clsdict.pop("__dict__", None) 
    336            clsdict.pop("__weakref__", None) 
    337            cls = type(cls.__name__, cls.__bases__, clsdict) 
    338            if constructor is not None: 
    339                constructor_fn = clsdict[constructor] 
    340 
    341        else: 
    342            cls.__doc__ = doc 
    343            if constructor is not None: 
    344                constructor_fn = getattr(cls, constructor) 
    345 
    346        if constructor is not None: 
    347            assert constructor_fn is not None 
    348            assert wtype is not None 
    349            setattr( 
    350                cls, 
    351                constructor, 
    352                _decorate_with_warning( 
    353                    constructor_fn, wtype, message, version, None 
    354                ), 
    355            ) 
    356    return cls 
    357 
    358 
    359def _decorate_with_warning( 
    360    func: _F, 
    361    wtype: Type[exc.SADeprecationWarning], 
    362    message: str, 
    363    version: str, 
    364    docstring_header: Optional[str] = None, 
    365    enable_warnings: bool = True, 
    366) -> _F: 
    367    """Wrap a function with a warnings.warn and augmented docstring.""" 
    368 
    369    message = _sanitize_restructured_text(message) 
    370 
    371    if issubclass(wtype, exc.Base20DeprecationWarning): 
    372        doc_only = ( 
    373            " (Background on SQLAlchemy 2.0 at: " 
    374            ":ref:`migration_20_toplevel`)" 
    375        ) 
    376    else: 
    377        doc_only = "" 
    378 
    379    @decorator 
    380    def warned(fn: _F, *args: Any, **kwargs: Any) -> _F: 
    381        skip_warning = not enable_warnings or kwargs.pop( 
    382            "_sa_skip_warning", False 
    383        ) 
    384        if not skip_warning: 
    385            _warn_with_version(message, version, wtype, stacklevel=3) 
    386        return fn(*args, **kwargs)  # type: ignore[no-any-return] 
    387 
    388    doc = func.__doc__ is not None and func.__doc__ or "" 
    389    if docstring_header is not None: 
    390        docstring_header %= dict(func=func.__name__) 
    391 
    392        docstring_header += doc_only 
    393 
    394        doc = inject_docstring_text(doc, docstring_header, 1) 
    395 
    396    decorated = warned(func) 
    397    decorated.__doc__ = doc 
    398    decorated._sa_warn = lambda: _warn_with_version(  # type: ignore 
    399        message, version, wtype, stacklevel=3 
    400    ) 
    401    return decorated