Issue #25609: Introduce contextlib.AbstractContextManager and

typing.ContextManager.

4 files changed, 89 insertions, 11 deletions

diff --git a/Lib/contextlib.py b/Lib/contextlib.py
index 5377987..98903bc 100644
--- a/Lib/contextlib.py
+++ b/Lib/contextlib.py
@@ -1,11 +1,34 @@
 """Utilities for with-statement contexts.  See PEP 343."""
-
+import abc
 import sys
 from collections import deque
 from functools import wraps
 
-__all__ = ["contextmanager", "closing", "ContextDecorator", "ExitStack",
-           "redirect_stdout", "redirect_stderr", "suppress"]
+__all__ = ["contextmanager", "closing", "AbstractContextManager",
+           "ContextDecorator", "ExitStack", "redirect_stdout",
+           "redirect_stderr", "suppress"]
+
+
+class AbstractContextManager(abc.ABC):
+
+    """An abstract base class for context managers."""
+
+    def __enter__(self):
+        """Return `self` upon entering the runtime context."""
+        return self
+
+    @abc.abstractmethod
+    def __exit__(self, exc_type, exc_value, traceback):
+        """Raise any exception triggered within the runtime context."""
+        return None
+
+    @classmethod
+    def __subclasshook__(cls, C):
+        if cls is AbstractContextManager:
+            if (any("__enter__" in B.__dict__ for B in C.__mro__) and
+                any("__exit__" in B.__dict__ for B in C.__mro__)):
+                    return True
+        return NotImplemented
 
 
 class ContextDecorator(object):
@@ -31,7 +54,7 @@ class ContextDecorator(object):
         return inner
 
 
-class _GeneratorContextManager(ContextDecorator):
+class _GeneratorContextManager(ContextDecorator, AbstractContextManager):
     """Helper for @contextmanager decorator."""
 
     def __init__(self, func, args, kwds):
@@ -134,7 +157,7 @@ def contextmanager(func):
     return helper
 
 
-class closing(object):
+class closing(AbstractContextManager):
     """Context to automatically close something at the end of a block.
 
     Code like this:
@@ -159,7 +182,7 @@ class closing(object):
         self.thing.close()
 
 
-class _RedirectStream:
+class _RedirectStream(AbstractContextManager):
 
     _stream = None
 
@@ -199,7 +222,7 @@ class redirect_stderr(_RedirectStream):
     _stream = "stderr"
 
 
-class suppress:
+class suppress(AbstractContextManager):
     """Context manager to suppress specified exceptions
 
     After the exception is suppressed, execution proceeds with the next
@@ -230,7 +253,7 @@ class suppress:
 
 
 # Inspired by discussions on http://bugs.python.org/issue13585
-class ExitStack(object):
+class ExitStack(AbstractContextManager):
     """Context manager for dynamic management of a stack of exit callbacks
 
     For example:
@@ -309,9 +332,6 @@ class ExitStack(object):
         """Immediately unwind the context stack"""
         self.__exit__(None, None, None)
 
-    def __enter__(self):
-        return self
-
     def __exit__(self, *exc_details):
         received_exc = exc_details[0] is not None
 
diff --git a/Lib/test/test_contextlib.py b/Lib/test/test_contextlib.py
index 04fc875..5c8bc98 100644
--- a/Lib/test/test_contextlib.py
+++ b/Lib/test/test_contextlib.py
@@ -12,6 +12,39 @@ except ImportError:
     threading = None
 
 
+class TestAbstractContextManager(unittest.TestCase):
+
+    def test_enter(self):
+        class DefaultEnter(AbstractContextManager):
+            def __exit__(self, *args):
+                super().__exit__(*args)
+
+        manager = DefaultEnter()
+        self.assertIs(manager.__enter__(), manager)
+
+    def test_exit_is_abstract(self):
+        class MissingExit(AbstractContextManager):
+            pass
+
+        with self.assertRaises(TypeError):
+            MissingExit()
+
+    def test_structural_subclassing(self):
+        class ManagerFromScratch:
+            def __enter__(self):
+                return self
+            def __exit__(self, exc_type, exc_value, traceback):
+                return None
+
+        self.assertTrue(issubclass(ManagerFromScratch, AbstractContextManager))
+
+        class DefaultEnter(AbstractContextManager):
+            def __exit__(self, *args):
+                super().__exit__(*args)
+
+        self.assertTrue(issubclass(DefaultEnter, AbstractContextManager))
+
+
 class ContextManagerTestCase(unittest.TestCase):
 
     def test_contextmanager_plain(self):
diff --git a/Lib/test/test_typing.py b/Lib/test/test_typing.py
index f1c6e12..a360824 100644
--- a/Lib/test/test_typing.py
+++ b/Lib/test/test_typing.py
@@ -1,3 +1,4 @@
+import contextlib
 import pickle
 import re
 import sys
@@ -1309,6 +1310,21 @@ class CollectionsAbcTests(TestCase):
         assert len(MMB[KT, VT]()) == 0
 
 
+class OtherABCTests(TestCase):
+
+    @skipUnless(hasattr(typing, 'ContextManager'),
+                'requires typing.ContextManager')
+    def test_contextmanager(self):
+        @contextlib.contextmanager
+        def manager():
+            yield 42
+
+        cm = manager()
+        assert isinstance(cm, typing.ContextManager)
+        assert isinstance(cm, typing.ContextManager[int])
+        assert not isinstance(42, typing.ContextManager)
+
+
 class NamedTupleTests(TestCase):
 
     def test_basics(self):
@@ -1447,6 +1463,8 @@ class AllTests(TestCase):
         assert 'ValuesView' in a
         assert 'cast' in a
         assert 'overload' in a
+        if hasattr(contextlib, 'AbstractContextManager'):
+            assert 'ContextManager' in a
         # Check that io and re are not exported.
         assert 'io' not in a
         assert 're' not in a
diff --git a/Lib/typing.py b/Lib/typing.py
index 6ead3c4..42a9ea3 100644
--- a/Lib/typing.py
+++ b/Lib/typing.py
@@ -1,6 +1,7 @@
 import abc
 from abc import abstractmethod, abstractproperty
 import collections
+import contextlib
 import functools
 import re as stdlib_re  # Avoid confusion with the re we export.
 import sys
@@ -1530,6 +1531,12 @@ class ValuesView(MappingView[VT_co], extra=collections_abc.ValuesView):
     pass
 
 
+if hasattr(contextlib, 'AbstractContextManager'):
+    class ContextManager(Generic[T_co], extra=contextlib.AbstractContextManager):
+        __slots__ = ()
+    __all__.append('ContextManager')
+
+
 class Dict(dict, MutableMapping[KT, VT]):
 
     def __new__(cls, *args, **kwds):