add missing __class_getitem__ from cpython 3.10

Signed-off-by: snowapril <sinjihng@gmail.com>

Lib/_collections_abc.py

@@ -9,6 +9,12 @@ Unit tests are in test_collections.
 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 @@ class Awaitable(metaclass=ABCMeta):
             return _check_methods(C, "__await__")
         return NotImplemented
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 
 class Coroutine(Awaitable):
 
@@ -169,6 +177,8 @@ class AsyncIterable(metaclass=ABCMeta):
             return _check_methods(C, "__aiter__")
         return NotImplemented
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 
 class AsyncIterator(AsyncIterable):
 
@@ -255,6 +265,8 @@ class Iterable(metaclass=ABCMeta):
             return _check_methods(C, "__iter__")
         return NotImplemented
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 
 class Iterator(Iterable):
 
@@ -384,6 +396,8 @@ class Container(metaclass=ABCMeta):
             return _check_methods(C, "__contains__")
         return NotImplemented
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 class Collection(Sized, Iterable, Container):
 
     __slots__ = ()
@@ -394,6 +408,141 @@ class Collection(Sized, Iterable, Container):
             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 @@ class Callable(metaclass=ABCMeta):
             return _check_methods(C, "__call__")
         return NotImplemented
 
+    __class_getitem__ = classmethod(_CallableGenericAlias)
+
 
 ### SETS ###
 
@@ -703,6 +854,8 @@ class MappingView(Sized):
     def __repr__(self):
         return '{0.__class__.__name__}({0._mapping!r})'.format(self)
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 
 class KeysView(MappingView, Set):
 

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 @@ class WeakSet:
 
     def __repr__(self):
         return repr(self.data)
+
+    __class_getitem__ = classmethod(GenericAlias)

Lib/asyncio/futures.py

@@ -184,6 +184,9 @@ class Future:
                 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.
 

Lib/asyncio/queues.py

@@ -81,6 +81,9 @@ class Queue:
     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):

Lib/concurrent/futures/_base.py

@@ -7,6 +7,7 @@ import collections
 import logging
 import threading
 import time
+import types
 
 FIRST_COMPLETED = 'FIRST_COMPLETED'
 FIRST_EXCEPTION = 'FIRST_EXCEPTION'
@@ -544,6 +545,8 @@ class Future(object):
             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."""
 

Lib/concurrent/futures/thread.py

@@ -10,6 +10,7 @@ from concurrent.futures import _base
 import itertools
 import queue
 import threading
+import types
 import weakref
 import os
 
@@ -62,6 +63,8 @@ class _WorkItem(object):
         else:
             self.future.set_result(result)
 
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 
 def _worker(executor_reference, work_queue, initializer, initargs):
     if initializer is not None:

Lib/dataclasses.py

@@ -7,6 +7,7 @@ import keyword
 import builtins
 import functools
 import _thread
+from types import GenericAlias
 
 
 __all__ = ['dataclass',
@@ -217,6 +218,8 @@ class InitVar(metaclass=_InitVarMeta):
             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 @@ class Field:
             # it.
             func(self.default, owner, name)
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 
 class _DataclassParams:
     __slots__ = ('init',

Lib/difflib.py

@@ -32,6 +32,7 @@ __all__ = ['get_close_matches', 'ndiff', 'restore', 'SequenceMatcher',
 
 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 @@ class SequenceMatcher:
         # 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.
 

Lib/functools.py

@@ -22,6 +22,7 @@ try:
     from _thread import RLock
 except ModuleNotFoundError:
     from _dummy_thread import RLock
+from types import GenericAlias
 
 
 ################################################################################
@@ -427,6 +428,8 @@ class partialmethod(object):
     def __isabstractmethod__(self):
         return getattr(self.func, "__isabstractmethod__", False)
 
+    __class_getitem__ = classmethod(GenericAlias)
+
 # Helper functions
 
 def _unwrap_partial(func):
@@ -977,3 +980,5 @@ class cached_property:
                         )
                         raise TypeError(msg) from None
         return val
+
+    __class_getitem__ = classmethod(GenericAlias)

Lib/multiprocessing/managers.py

@@ -18,6 +18,7 @@ import sys
 import threading
 import array
 import queue
+import types
 
 from time import time as _time
 from traceback import format_exc
@@ -1078,6 +1079,8 @@ class ValueProxy(BaseProxy):
         return self._callmethod('set', (value,))
     value = property(get, set)
 
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 
 BaseListProxy = MakeProxyType('BaseListProxy', (
     '__add__', '__contains__', '__delitem__', '__getitem__', '__len__',

Lib/multiprocessing/pool.py

@@ -20,6 +20,7 @@ import collections
 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 @@ class ApplyResult(object):
         self._event.set()
         del self._cache[self._job]
 
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 AsyncResult = ApplyResult       # create alias -- see #17805
 
 #

Lib/multiprocessing/queues.py

@@ -14,6 +14,7 @@ import os
 import threading
 import collections
 import time
+import types
 import weakref
 import errno
 
@@ -353,3 +354,5 @@ class SimpleQueue(object):
         else:
             with self._wlock:
                 self._writer.send_bytes(obj)
+
+    __class_getitem__ = classmethod(types.GenericAlias)

Lib/queue.py

@@ -1,6 +1,7 @@
 '''A multi-producer, multi-consumer queue.'''
 
 import threading
+import types
 from collections import deque
 from heapq import heappush, heappop
 from time import monotonic as time
@@ -216,6 +217,8 @@ class Queue:
     def _get(self):
         return self.queue.popleft()
 
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 
 class PriorityQueue(Queue):
     '''Variant of Queue that retrieves open entries in priority order (lowest first).
@@ -316,6 +319,8 @@ class _PySimpleQueue:
         '''Return the approximate size of the queue (not reliable!).'''
         return len(self._queue)
 
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 
 if SimpleQueue is None:
     SimpleQueue = _PySimpleQueue

Lib/tempfile.py

@@ -44,6 +44,7 @@ import shutil as _shutil
 import errno as _errno
 from random import Random as _Random
 import sys as _sys
+import types as _types
 import weakref as _weakref
 
 try:
@@ -646,6 +647,8 @@ class SpooledTemporaryFile:
                                    'encoding': encoding, 'newline': newline,
                                    'dir': dir, 'errors': errors}
 
+    __class_getitem__ = classmethod(_types.GenericAlias)
+
     def _check(self, file):
         if self._rolled: return
         max_size = self._max_size
@@ -833,3 +836,5 @@ class TemporaryDirectory(object):
     def cleanup(self):
         if self._finalizer.detach():
             self._rmtree(self.name)
+
+    __class_getitem__ = classmethod(_types.GenericAlias)

Lib/unittest/case.py

@@ -10,6 +10,7 @@ import warnings
 import collections
 import contextlib
 import traceback
+import types
 
 from . import result
 from .util import (strclass, safe_repr, _count_diff_all_purpose,
@@ -217,6 +218,8 @@ class _AssertRaisesContext(_AssertRaisesBaseContext):
                      expected_regex.pattern, str(exc_value)))
         return True
 
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 
 class _AssertWarnsContext(_AssertRaisesBaseContext):
     """A context manager used to implement TestCase.assertWarns* methods."""