mercurial/thirdparty/attr/_compat.py
author Matt Harbison <matt_harbison@yahoo.com>
Mon, 21 Nov 2022 15:04:42 -0500
changeset 49643 e1c586b9a43c
parent 34397 765eb17a7eb8
permissions -rw-r--r--
attr: vendor 22.1.0 The previous version was 5 years old, and pytype 2022.06.30 started complaining about various uses (e.g. seeing `mercurial.thirdparty.attr._make._CountingAttr` instead of `bytearray`). Hopefully this helps. Additionally, this has official python 3.11 support. The `attrs` package is left out, because it is simply a bunch of *.pyi stubs and `from attr.X import *`, and that's not how they've been used up to this point. We'd probably need to customize those anyway to `from mercurial.thirdparty.attr import *`.

# SPDX-License-Identifier: MIT


import inspect
import platform
import sys
import threading
import types
import warnings

from collections.abc import Mapping, Sequence  # noqa


PYPY = platform.python_implementation() == "PyPy"
PY36 = sys.version_info[:2] >= (3, 6)
HAS_F_STRINGS = PY36
PY310 = sys.version_info[:2] >= (3, 10)


if PYPY or PY36:
    ordered_dict = dict
else:
    from collections import OrderedDict

    ordered_dict = OrderedDict


def just_warn(*args, **kw):
    warnings.warn(
        "Running interpreter doesn't sufficiently support code object "
        "introspection.  Some features like bare super() or accessing "
        "__class__ will not work with slotted classes.",
        RuntimeWarning,
        stacklevel=2,
    )


class _AnnotationExtractor:
    """
    Extract type annotations from a callable, returning None whenever there
    is none.
    """

    __slots__ = ["sig"]

    def __init__(self, callable):
        try:
            self.sig = inspect.signature(callable)
        except (ValueError, TypeError):  # inspect failed
            self.sig = None

    def get_first_param_type(self):
        """
        Return the type annotation of the first argument if it's not empty.
        """
        if not self.sig:
            return None

        params = list(self.sig.parameters.values())
        if params and params[0].annotation is not inspect.Parameter.empty:
            return params[0].annotation

        return None

    def get_return_type(self):
        """
        Return the return type if it's not empty.
        """
        if (
            self.sig
            and self.sig.return_annotation is not inspect.Signature.empty
        ):
            return self.sig.return_annotation

        return None


def make_set_closure_cell():
    """Return a function of two arguments (cell, value) which sets
    the value stored in the closure cell `cell` to `value`.
    """
    # pypy makes this easy. (It also supports the logic below, but
    # why not do the easy/fast thing?)
    if PYPY:

        def set_closure_cell(cell, value):
            cell.__setstate__((value,))

        return set_closure_cell

    # Otherwise gotta do it the hard way.

    # Create a function that will set its first cellvar to `value`.
    def set_first_cellvar_to(value):
        x = value
        return

        # This function will be eliminated as dead code, but
        # not before its reference to `x` forces `x` to be
        # represented as a closure cell rather than a local.
        def force_x_to_be_a_cell():  # pragma: no cover
            return x

    try:
        # Extract the code object and make sure our assumptions about
        # the closure behavior are correct.
        co = set_first_cellvar_to.__code__
        if co.co_cellvars != ("x",) or co.co_freevars != ():
            raise AssertionError  # pragma: no cover

        # Convert this code object to a code object that sets the
        # function's first _freevar_ (not cellvar) to the argument.
        if sys.version_info >= (3, 8):

            def set_closure_cell(cell, value):
                cell.cell_contents = value

        else:
            args = [co.co_argcount]
            args.append(co.co_kwonlyargcount)
            args.extend(
                [
                    co.co_nlocals,
                    co.co_stacksize,
                    co.co_flags,
                    co.co_code,
                    co.co_consts,
                    co.co_names,
                    co.co_varnames,
                    co.co_filename,
                    co.co_name,
                    co.co_firstlineno,
                    co.co_lnotab,
                    # These two arguments are reversed:
                    co.co_cellvars,
                    co.co_freevars,
                ]
            )
            set_first_freevar_code = types.CodeType(*args)

            def set_closure_cell(cell, value):
                # Create a function using the set_first_freevar_code,
                # whose first closure cell is `cell`. Calling it will
                # change the value of that cell.
                setter = types.FunctionType(
                    set_first_freevar_code, {}, "setter", (), (cell,)
                )
                # And call it to set the cell.
                setter(value)

        # Make sure it works on this interpreter:
        def make_func_with_cell():
            x = None

            def func():
                return x  # pragma: no cover

            return func

        cell = make_func_with_cell().__closure__[0]
        set_closure_cell(cell, 100)
        if cell.cell_contents != 100:
            raise AssertionError  # pragma: no cover

    except Exception:
        return just_warn
    else:
        return set_closure_cell


set_closure_cell = make_set_closure_cell()

# Thread-local global to track attrs instances which are already being repr'd.
# This is needed because there is no other (thread-safe) way to pass info
# about the instances that are already being repr'd through the call stack
# in order to ensure we don't perform infinite recursion.
#
# For instance, if an instance contains a dict which contains that instance,
# we need to know that we're already repr'ing the outside instance from within
# the dict's repr() call.
#
# This lives here rather than in _make.py so that the functions in _make.py
# don't have a direct reference to the thread-local in their globals dict.
# If they have such a reference, it breaks cloudpickle.
repr_context = threading.local()