added run_when_generator_completes and RunActionAfterGeneratorCompletes

CHANGELOG.md

 # v2.26.1
 
-* _TBA_
+* added run_when_generator_completes and RunActionAfterGeneratorCompletes
 
 # v2.26.0
 

docs/coding/sequences.rst

 Sequences and iterators
 #######################
 
+Generators
+==========
+
+.. autoclass:: satella.coding.RunActionAfterGeneratorCompletes
+    :members:
+
+.. autoclass:: satella.coding.
+
 Rolling averages
 ================
 

docs/conf.py

@@ -151,3 +151,11 @@ texinfo_documents = [
      author, 'satella', 'One line description of project.',
      'Miscellaneous'),
 ]
+
+autoclass_content = 'both'
+
+autodoc_default_options = {
+    'members': True,
+}
+autodoc_typehints = "description"
+autoclass_content = 'both'

satella/__init__.py

-__version__ = '2.26.1a1'
+__version__ = '2.26.1'

satella/coding/__init__.py

@@ -23,8 +23,10 @@ from .misc import update_if_not_none, update_key_if_none, update_attr_if_none, q
 from .overloading import overload, class_or_instancemethod, TypeSignature
 from .recast_exceptions import rethrow_as, silence_excs, catch_exception, log_exceptions, \
     raises_exception, reraise_as
+from .generators import RunActionAfterGeneratorCompletes, run_when_generator_completes
 
 __all__ = [
+    'RunActionAfterGeneratorCompletes', 'run_when_generator_completes',
     'EmptyContextManager', 'Context', 'length',
     'assert_equal', 'InequalityReason', 'Inequal', 'wrap_callable_in_context_manager',
     'Closeable', 'contains', 'enum_value',

satella/coding/generators.py

+import typing as tp
+from abc import ABCMeta, abstractmethod
+
+
+class RunActionAfterGeneratorCompletes(metaclass=ABCMeta):
+    """
+    Run an action after a generator completes.
+    An abstract class.
+    """
+
+    __slots__ = 'generator', 'args', 'kwargs'
+
+    def __init__(self, generator: tp.Generator, *args, **kwargs):
+        """
+        :param generator: generator to watch for
+        :param args: arguments to invoke action_to_run with
+        :param kwargs: keyword arguments to invoke action_to_run with
+        """
+        self.generator = generator
+        self.args = args
+        self.kwargs = kwargs
+
+    def send(self, value):
+        """Send a value to the generator"""
+        self.generator.send(value)
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        try:
+            return next(self.generator)
+        except StopIteration:
+            self.action_to_run(*self.args, **self.kwargs)
+            raise
+
+    @abstractmethod
+    def action_to_run(self):
+        """This will run when this generator completes. Override it."""
+
+
+def run_when_generator_completes(gen: tp.Generator, call_on_done: tp.Callable[[], None],
+                                 *args, **kwargs) -> tp.Generator:
+    """
+    Return the generator with call_on_done to be called on when it finishes
+
+    :param gen: generator
+    :param call_on_done: callable/0 to call on generator's completion
+    :param args: args to pass to the callable
+    :param kwargs: kwargs to pass to the callable
+    :returns: generator
+    """
+    class Inner(RunActionAfterGeneratorCompletes):
+        def action_to_run(self, *args, **kwargs):
+            call_on_done(*args, **kwargs)
+
+    return Inner(gen, *args, **kwargs)

tests/test_coding/test_iterators.py

 import sys
 import unittest
 
-from satella.coding import SelfClosingGenerator, hint_with_length, chain
+from satella.coding import SelfClosingGenerator, hint_with_length, chain, run_when_generator_completes
 from satella.coding.sequences import smart_enumerate, ConstruableIterator, walk, \
     IteratorListAdapter, is_empty, ListWrapperIterator
 
@@ -16,6 +16,26 @@ def iterate():
 
 class TestIterators(unittest.TestCase):
 
+    def test_run_when_generator_completes(self):
+        called = False
+
+        def generator():
+            yield 1
+            yield 2
+            yield 3
+
+        def mark_done(f):
+            assert f == 2
+            nonlocal called
+            called = True
+
+        gen = run_when_generator_completes(generator(), mark_done, 2)
+        a = next(gen)
+        self.assertFalse(called)
+        for i in gen:
+            pass
+        self.assertTrue(called)
+
     def test_list_wrapper_iterator_contains(self):
 
         lwe = ListWrapperIterator(iterate())