add missing __class_getitem__ from cpython 3.10

Snowapril · Snowapril · commit b2eff4ae53ba · 2021-10-22T18:43:15.000+09:00
Signed-off-by: snowapril &lt;sinjihng@gmail.com&gt;
diff --git a/Lib/_collections_abc.py b/Lib/_collections_abc.py
@@ -9,6 +9,12 @@
 from abc import ABCMeta, abstractmethod
 import sys
 
+GenericAlias = type(list[int])
+EllipsisType = type(...)
+def _f(): pass
+FunctionType = type(_f)
+del _f
+
 __all__ = ["Awaitable", "Coroutine",
            "AsyncIterable", "AsyncIterator", "AsyncGenerator",
            "Hashable", "Iterable", "Iterator", "Generator", "Reversible",
@@ -110,6 +116,8 @@ def __subclasshook__(cls, C):
             return _check_methods(C, "__await__")
         return NotImplemented
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 
 class Coroutine(Awaitable):
 
@@ -169,6 +177,8 @@ def __subclasshook__(cls, C):
             return _check_methods(C, "__aiter__")
         return NotImplemented
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 
 class AsyncIterator(AsyncIterable):
 
@@ -255,6 +265,8 @@ def __subclasshook__(cls, C):
             return _check_methods(C, "__iter__")
         return NotImplemented
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 
 class Iterator(Iterable):
 
@@ -384,6 +396,8 @@ def __subclasshook__(cls, C):
             return _check_methods(C, "__contains__")
         return NotImplemented
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 class Collection(Sized, Iterable, Container):
 
     __slots__ = ()
@@ -394,6 +408,141 @@ def __subclasshook__(cls, C):
             return _check_methods(C,  "__len__", "__iter__", "__contains__")
         return NotImplemented
 
+
+class _CallableGenericAlias(GenericAlias):
+    """ Represent `Callable[argtypes, resulttype]`.
+
+    This sets ``__args__`` to a tuple containing the flattened ``argtypes``
+    followed by ``resulttype``.
+
+    Example: ``Callable[[int, str], float]`` sets ``__args__`` to
+    ``(int, str, float)``.
+    """
+
+    __slots__ = ()
+
+    def __new__(cls, origin, args):
+        if not (isinstance(args, tuple) and len(args) == 2):
+            raise TypeError(
+                "Callable must be used as Callable[[arg, ...], result].")
+        t_args, t_result = args
+        if isinstance(t_args, list):
+            args = (*t_args, t_result)
+        elif not _is_param_expr(t_args):
+            raise TypeError(f"Expected a list of types, an ellipsis, "
+                            f"ParamSpec, or Concatenate. Got {t_args}")
+        return super().__new__(cls, origin, args)
+
+    @property
+    def __parameters__(self):
+        params = []
+        for arg in self.__args__:
+            # Looks like a genericalias
+            if hasattr(arg, "__parameters__") and isinstance(arg.__parameters__, tuple):
+                params.extend(arg.__parameters__)
+            else:
+                if _is_typevarlike(arg):
+                    params.append(arg)
+        return tuple(dict.fromkeys(params))
+
+    def __repr__(self):
+        if len(self.__args__) == 2 and _is_param_expr(self.__args__[0]):
+            return super().__repr__()
+        return (f'collections.abc.Callable'
+                f'[[{", ".join([_type_repr(a) for a in self.__args__[:-1]])}], '
+                f'{_type_repr(self.__args__[-1])}]')
+
+    def __reduce__(self):
+        args = self.__args__
+        if not (len(args) == 2 and _is_param_expr(args[0])):
+            args = list(args[:-1]), args[-1]
+        return _CallableGenericAlias, (Callable, args)
+
+    def __getitem__(self, item):
+        # Called during TypeVar substitution, returns the custom subclass
+        # rather than the default types.GenericAlias object.  Most of the
+        # code is copied from typing's _GenericAlias and the builtin
+        # types.GenericAlias.
+
+        # A special case in PEP 612 where if X = Callable[P, int],
+        # then X[int, str] == X[[int, str]].
+        param_len = len(self.__parameters__)
+        if param_len == 0:
+            raise TypeError(f'{self} is not a generic class')
+        if not isinstance(item, tuple):
+            item = (item,)
+        if (param_len == 1 and _is_param_expr(self.__parameters__[0])
+                and item and not _is_param_expr(item[0])):
+            item = (list(item),)
+        item_len = len(item)
+        if item_len != param_len:
+            raise TypeError(f'Too {"many" if item_len > param_len else "few"}'
+                            f' arguments for {self};'
+                            f' actual {item_len}, expected {param_len}')
+        subst = dict(zip(self.__parameters__, item))
+        new_args = []
+        for arg in self.__args__:
+            if _is_typevarlike(arg):
+                if _is_param_expr(arg):
+                    arg = subst[arg]
+                    if not _is_param_expr(arg):
+                        raise TypeError(f"Expected a list of types, an ellipsis, "
+                                        f"ParamSpec, or Concatenate. Got {arg}")
+                else:
+                    arg = subst[arg]
+            # Looks like a GenericAlias
+            elif hasattr(arg, '__parameters__') and isinstance(arg.__parameters__, tuple):
+                subparams = arg.__parameters__
+                if subparams:
+                    subargs = tuple(subst[x] for x in subparams)
+                    arg = arg[subargs]
+            new_args.append(arg)
+
+        # args[0] occurs due to things like Z[[int, str, bool]] from PEP 612
+        if not isinstance(new_args[0], list):
+            t_result = new_args[-1]
+            t_args = new_args[:-1]
+            new_args = (t_args, t_result)
+        return _CallableGenericAlias(Callable, tuple(new_args))
+
+
+def _is_typevarlike(arg):
+    obj = type(arg)
+    # looks like a TypeVar/ParamSpec
+    return (obj.__module__ == 'typing'
+            and obj.__name__ in {'ParamSpec', 'TypeVar'})
+
+def _is_param_expr(obj):
+    """Checks if obj matches either a list of types, ``...``, ``ParamSpec`` or
+    ``_ConcatenateGenericAlias`` from typing.py
+    """
+    if obj is Ellipsis:
+        return True
+    if isinstance(obj, list):
+        return True
+    obj = type(obj)
+    names = ('ParamSpec', '_ConcatenateGenericAlias')
+    return obj.__module__ == 'typing' and any(obj.__name__ == name for name in names)
+
+def _type_repr(obj):
+    """Return the repr() of an object, special-casing types (internal helper).
+
+    Copied from :mod:`typing` since collections.abc
+    shouldn't depend on that module.
+    """
+    if isinstance(obj, GenericAlias):
+        return repr(obj)
+    if isinstance(obj, type):
+        if obj.__module__ == 'builtins':
+            return obj.__qualname__
+        return f'{obj.__module__}.{obj.__qualname__}'
+    if obj is Ellipsis:
+        return '...'
+    if isinstance(obj, FunctionType):
+        return obj.__name__
+    return repr(obj)
+
+
 class Callable(metaclass=ABCMeta):
 
     __slots__ = ()
@@ -408,6 +557,8 @@ def __subclasshook__(cls, C):
             return _check_methods(C, "__call__")
         return NotImplemented
 
+    __class_getitem__ = classmethod(_CallableGenericAlias)
+
 
 ### SETS ###
 
@@ -703,6 +854,8 @@ def __len__(self):
     def __repr__(self):
         return '{0.__class__.__name__}({0._mapping!r})'.format(self)
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 
 class KeysView(MappingView, Set):
 
diff --git a/Lib/_weakrefset.py b/Lib/_weakrefset.py
@@ -3,6 +3,7 @@
 # by abc.py to load everything else at startup.
 
 from _weakref import ref
+from types import GenericAlias
 
 __all__ = ['WeakSet']
 
@@ -197,3 +198,5 @@ def isdisjoint(self, other):
 
     def __repr__(self):
         return repr(self.data)
+
+    __class_getitem__ = classmethod(GenericAlias)
diff --git a/Lib/asyncio/futures.py b/Lib/asyncio/futures.py
@@ -184,6 +184,9 @@ def __del__(self):
                 context['source_traceback'] = self._source_traceback
             self._loop.call_exception_handler(context)
 
+    def __class_getitem__(cls, type):
+        return cls
+
     def cancel(self):
         """Cancel the future and schedule callbacks.
 
diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py
@@ -81,6 +81,9 @@ def __repr__(self):
     def __str__(self):
         return '<{} {}>'.format(type(self).__name__, self._format())
 
+    def __class_getitem__(cls, type):
+        return cls
+
     def _format(self):
         result = 'maxsize={!r}'.format(self._maxsize)
         if getattr(self, '_queue', None):
diff --git a/Lib/concurrent/futures/_base.py b/Lib/concurrent/futures/_base.py
@@ -7,6 +7,7 @@
 import logging
 import threading
 import time
+import types
 
 FIRST_COMPLETED = 'FIRST_COMPLETED'
 FIRST_EXCEPTION = 'FIRST_EXCEPTION'
@@ -544,6 +545,8 @@ def set_exception(self, exception):
             self._condition.notify_all()
         self._invoke_callbacks()
 
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 class Executor(object):
     """This is an abstract base class for concrete asynchronous executors."""
 
diff --git a/Lib/concurrent/futures/thread.py b/Lib/concurrent/futures/thread.py
@@ -10,6 +10,7 @@
 import itertools
 import queue
 import threading
+import types
 import weakref
 import os
 
@@ -62,6 +63,8 @@ def run(self):
         else:
             self.future.set_result(result)
 
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 
 def _worker(executor_reference, work_queue, initializer, initargs):
     if initializer is not None:
diff --git a/Lib/dataclasses.py b/Lib/dataclasses.py
@@ -7,6 +7,7 @@
 import builtins
 import functools
 import _thread
+from types import GenericAlias
 
 
 __all__ = ['dataclass',
@@ -217,6 +218,8 @@ def __repr__(self):
             type_name = repr(self.type)
         return f'dataclasses.InitVar[{type_name}]'
 
+    def __class_getitem__(cls, type):
+        return InitVar(type)
 
 # Instances of Field are only ever created from within this module,
 # and only from the field() function, although Field instances are
@@ -285,6 +288,8 @@ def __set_name__(self, owner, name):
             # it.
             func(self.default, owner, name)
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 
 class _DataclassParams:
     __slots__ = ('init',
diff --git a/Lib/difflib.py b/Lib/difflib.py
@@ -32,6 +32,7 @@
 
 from heapq import nlargest as _nlargest
 from collections import namedtuple as _namedtuple
+from types import GenericAlias
 
 Match = _namedtuple('Match', 'a b size')
 
@@ -685,6 +686,8 @@ def real_quick_ratio(self):
         # shorter sequence
         return _calculate_ratio(min(la, lb), la + lb)
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 def get_close_matches(word, possibilities, n=3, cutoff=0.6):
     """Use SequenceMatcher to return list of the best "good enough" matches.
 
diff --git a/Lib/functools.py b/Lib/functools.py
@@ -22,6 +22,7 @@
     from _thread import RLock
 except ModuleNotFoundError:
     from _dummy_thread import RLock
+from types import GenericAlias
 
 
 ################################################################################
@@ -427,6 +428,8 @@ def __get__(self, obj, cls=None):
     def __isabstractmethod__(self):
         return getattr(self.func, "__isabstractmethod__", False)
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 # Helper functions
 
 def _unwrap_partial(func):
@@ -977,3 +980,5 @@ def __get__(self, instance, owner=None):
                         )
                         raise TypeError(msg) from None
         return val
+
+    __class_getitem__ = classmethod(GenericAlias)
diff --git a/Lib/multiprocessing/managers.py b/Lib/multiprocessing/managers.py
@@ -18,6 +18,7 @@
 import threading
 import array
 import queue
+import types
 
 from time import time as _time
 from traceback import format_exc
@@ -1078,6 +1079,8 @@ def set(self, value):
         return self._callmethod('set', (value,))
     value = property(get, set)
 
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 
 BaseListProxy = MakeProxyType('BaseListProxy', (
     '__add__', '__contains__', '__delitem__', '__getitem__', '__len__',
diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py
@@ -20,6 +20,7 @@
 import os
 import time
 import traceback
+import types
 
 # If threading is available then ThreadPool should be provided.  Therefore
 # we avoid top-level imports which are liable to fail on some systems.
@@ -616,6 +617,8 @@ def _set(self, i, obj):
         self._event.set()
         del self._cache[self._job]
 
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 AsyncResult = ApplyResult       # create alias -- see #17805
 
 #
diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py
@@ -14,6 +14,7 @@
 import threading
 import collections
 import time
+import types
 import weakref
 import errno
 
@@ -353,3 +354,5 @@ def put(self, obj):
         else:
             with self._wlock:
                 self._writer.send_bytes(obj)
+
+    __class_getitem__ = classmethod(types.GenericAlias)
diff --git a/Lib/queue.py b/Lib/queue.py
diff --git a/Lib/tempfile.py b/Lib/tempfile.py
diff --git a/Lib/unittest/case.py b/Lib/unittest/case.py
