implemented the transaction-manager run method

transaction/_manager.py

@@ -160,6 +160,42 @@ def _retryable(self, error_type, error):
             if (should_retry is not None) and should_retry(error):
                 return True
 
+    run_no_func_types = int, type(None)
+    def run(self, func=None, tries=3):
+        if isinstance(func, self.run_no_func_types):
+            if func is not None:
+                tries = func
+            return lambda func: self.run(func, tries)
+
+        if tries <= 0:
+            raise ValueError("tries must be positive")
+
+        name = func.__name__
+        doc = func.__doc__
+        if name != '_':
+            if doc:
+                doc = name + '\n\n' + doc
+            else:
+                doc = name
+
+        for i in range(1, tries + 1):
+            txn = self.begin()
+            if doc:
+                txn.note(doc)
+
+            try:
+                result = func()
+                txn.commit()
+            except Exception as v:
+                if i == tries:
+                    raise # that was our last chance
+                retry = self._retryable(v.__class__, v)
+                txn.abort()
+                if not retry:
+                    raise
+            else:
+                return result
+
 
 class ThreadTransactionManager(TransactionManager, threading.local):
     """Thread-aware transaction manager.

transaction/tests/test__manager.py

@@ -319,6 +319,121 @@ def test_attempts_w_default_count(self):
             self.assertTrue(attempt.manager is tm)
         self.assertTrue(found[-1] is tm)
 
+    def test_run(self):
+        import transaction.interfaces
+        class Retry(transaction.interfaces.TransientError):
+            pass
+
+        tm = self._makeOne()
+        i = [0, None]
+
+        @tm.run()
+        def meaning():
+            "Nice doc"
+            i[0] += 1
+            i[1] = tm.get()
+            if i[0] < 3:
+                raise Retry
+            return 42
+
+        self.assertEqual(i[0], 3)
+        self.assertEqual(meaning, 42)
+        self.assertEqual(i[1].description, "meaning\n\nNice doc")
+
+    def test_run_no_name_explicit_tries(self):
+        import transaction.interfaces
+        class Retry(transaction.interfaces.TransientError):
+            pass
+
+        tm = self._makeOne()
+        i = [0, None]
+
+        @tm.run(4)
+        def _():
+            "Nice doc"
+            i[0] += 1
+            i[1] = tm.get()
+            if i[0] < 4:
+                raise Retry
+
+        self.assertEqual(i[0], 4)
+        self.assertEqual(i[1].description, "Nice doc")
+
+    def test_run_pos_tries(self):
+        tm = self._makeOne()
+
+        with self.assertRaises(ValueError):
+            tm.run(0)(lambda : None)
+        with self.assertRaises(ValueError):
+            @tm.run(-1)
+            def _():
+                pass
+
+    def test_run_stop_on_success(self):
+        import transaction.interfaces
+        class Retry(transaction.interfaces.TransientError):
+            pass
+
+        tm = self._makeOne()
+        i = [0, None]
+
+        @tm.run()
+        def meaning():
+            i[0] += 1
+            i[1] = tm.get()
+            return 43
+
+        self.assertEqual(i[0], 1)
+        self.assertEqual(meaning, 43)
+        self.assertEqual(i[1].description, "meaning")
+
+    def test_run_retries_but_gives_up(self):
+        import transaction.interfaces
+        class Retry(transaction.interfaces.TransientError):
+            pass
+
+        tm = self._makeOne()
+        i = [0]
+
+        with self.assertRaises(Retry):
+            @tm.run()
+            def _():
+                i[0] += 1
+                raise Retry
+
+        self.assertEqual(i[0], 3)
+
+    def test_run_propigates_errors(self):
+        tm = self._makeOne()
+        with self.assertRaises(ValueError):
+            @tm.run
+            def _():
+                raise ValueError
+
+    def test_run_defer_to_dm(self):
+        import transaction.tests.savepointsample
+
+        class DM(transaction.tests.savepointsample.SampleSavepointDataManager):
+            def should_retry(self, e):
+                if 'should retry' in str(e):
+                    return True
+
+        ntry = [0]
+        dm = transaction.tests.savepointsample.SampleSavepointDataManager()
+        dm2 = DM()
+        with transaction.manager:
+            dm2['ntry'] = 0
+
+        @transaction.manager.run
+        def _():
+            ntry[0] += 1
+            dm['ntry'] = ntry[0]
+            dm2['ntry'] = ntry[0]
+            if ntry[0] % 3:
+                raise ValueError('we really should retry this')
+
+        self.assertEqual(ntry[0], 3)
+
     def test__retryable_w_transient_error(self):
         from transaction.interfaces import TransientError
         tm = self._makeOne()