loop_n.py

import attr
from twisted.internet import defer, task


@attr.s
class LoopNTimes(object):
    _times = attr.ib()
    _f = attr.ib()
    _loopingCall = attr.ib(default=None, init=False)
    _called = attr.ib(default=0, init=False)

    @property
    def loopingCall(self):
        return self._loopingCall

    @loopingCall.setter
    def loopingCall(self, loopingCall):
        if self._loopingCall is not None:
            raise RuntimeError("looping call already set.")
        self._loopingCall = loopingCall

    def __call__(self, *args, **kwargs):
        if not self._loopingCall:
            raise RuntimeError(
                "You must associate me with LoopingCall"
                " by calling my setLoopingCall method")
        if self._called < (self._times - 1):
            self._called += 1
            return self._f(*args, **kwargs)
        else:
            d = defer.maybeDeferred(self._f, *args, **kwargs)

            @d.addBoth
            def stop(result):
                self._loopingCall.stop()
                return result
            return d



@task.react
def main(reactor):
    def aFunction(*args, **kwargs):
        print("Called with {} and {}".format(args, kwargs))
        return task.deferLater(reactor, 1.0, lambda: None)

    nTimes = LoopNTimes(1, aFunction)
    loop = task.LoopingCall(nTimes, 1, k=2)
    nTimes.loopingCall = loop
    return loop.start(5)