add `chain_callables`

CHANGELOG.md

@@ -4,3 +4,4 @@
 * added `ExecutorWrapper`
 * added `Future.on_success` and `Future.on_failure`
 * `apply_dict_object` won't DictObject if passed a DictObject already
+* added `chain_callables`

docs/coding/functions.rst

@@ -4,6 +4,8 @@ Functions and decorators
 
 .. autoclass:: satella.coding.class_or_instancemethod
 
+.. autofunction:: satella.coding.chain_callables
+
 .. autofunction:: satella.coding.source_to_function
 
 .. autofunction:: satella.coding.call_with_arguments

satella/__init__.py

-__version__ = '2.12.7_a9'
+__version__ = '2.12.7'

satella/coding/__init__.py

@@ -14,7 +14,7 @@ from .metaclasses import metaclass_maker, wrap_with, dont_wrap, wrap_property, D
     CopyDocsFrom
 from .misc import update_if_not_none, update_key_if_none, update_attr_if_none, queue_iterator, \
     update_key_if_not_none, source_to_function, update_key_if_true, \
-    get_arguments, call_with_arguments
+    get_arguments, call_with_arguments, chain_callables
 from .overloading import overload, class_or_instancemethod
 from .recast_exceptions import rethrow_as, silence_excs, catch_exception, log_exceptions, \
     raises_exception
@@ -25,7 +25,7 @@ __all__ = [
     'update_attr_if_none', 'update_key_if_not_none', 'source_to_function',
     'update_key_if_true',
     'hint_with_length', 'SelfClosingGenerator', 'exhaust', 'chain',
-    'Monitor', 'RMonitor', 'merge_dicts',
+    'Monitor', 'RMonitor', 'merge_dicts', 'chain_callables',
     'short_none', 'has_keys', 'CopyDocsFrom',
     'precondition', 'postcondition', 'queue_get',
     'rethrow_as', 'silence_excs', 'raises_exception',

satella/coding/misc.py

@@ -26,6 +26,29 @@ def update_if_not_none(dictionary: tp.Dict, key: tp.Hashable, value) -> tp.Dict:
     return update_key_if_none(dictionary, key, value)
 
 
+def chain_callables(callable1: tp.Callable, callable2: tp.Callable) -> tp.Callable:
+    """
+    Link two callables together. callable2, if it takes an argument, will receive
+    callables'1 result, and if it takes no arguments it will received nothing.
+
+    :param callable1: first callable to call
+    :param callable2: callable to call with callable1's result
+    :return: result of callable2
+    """
+    def inner(*args, **kwargs):
+        res = callable1(*args, **kwargs)
+        try:
+            res = callable2(res)
+        except TypeError as e:
+            if 'positional arguments but' in e.args[0] \
+                    and 'was given' in e.args[0] and 'takes' in e.args[0]:
+                res = callable2()
+            else:
+                raise
+        return res
+    return inner
+
+
 def source_to_function(src: str) -> tp.Callable[[tp.Any], tp.Any]:
     """
     Transform a string containing a Python expression with a variable x to a lambda.

tests/test_coding/test_misc.py

 import unittest
 
 from satella.coding import update_key_if_not_none, overload, class_or_instancemethod, \
-    update_key_if_true, get_arguments, call_with_arguments
+    update_key_if_true, get_arguments, call_with_arguments, chain_callables
 from satella.coding.structures import HashableMixin
 from satella.coding.transforms import jsonify, intify
 
 
 class TestCase(unittest.TestCase):
 
+    def test_chain_callables(self):
+        def a():
+            return 5
+
+        def mul2(b):
+            return b*2
+
+        self.assertEqual(chain_callables(a, mul2)(), 10)
+
+        def two():
+            return 2
+
+        self.assertEqual(chain_callables(a, two)(), 2)
+
     def test_hashable_mixin(self):
         class MixedIn(HashableMixin):
             pass