Python - Timer with asyncio/coroutine

Creating Task using ensure_future is a common way to start some job executing without blocking your execution flow. You can also cancel tasks.

I wrote example implementation for you to have something to start from:

import asyncio


class Timer:
    def __init__(self, timeout, callback):
        self._timeout = timeout
        self._callback = callback
        self._task = asyncio.ensure_future(self._job())

    async def _job(self):
        await asyncio.sleep(self._timeout)
        await self._callback()

    def cancel(self):
        self._task.cancel()


async def timeout_callback():
    await asyncio.sleep(0.1)
    print('echo!')


async def main():
    print('\nfirst example:')
    timer = Timer(2, timeout_callback)  # set timer for two seconds
    await asyncio.sleep(2.5)  # wait to see timer works

    print('\nsecond example:')
    timer = Timer(2, timeout_callback)  # set timer for two seconds
    await asyncio.sleep(1)
    timer.cancel()  # cancel it
    await asyncio.sleep(1.5)  # and wait to see it won't call callback


loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
    loop.run_until_complete(main())
finally:
    loop.run_until_complete(loop.shutdown_asyncgens())
    loop.close()

Output:

first example:
echo!

second example:

Thanks Mikhail Gerasimov for your answer, it was very useful. Here is an extension to Mikhail’s anwer. This is an interval timer with some twists. Perhaps it is useful for some users.

import asyncio


class Timer:
    def __init__(self, interval, first_immediately, timer_name, context, callback):
        self._interval = interval
        self._first_immediately = first_immediately
        self._name = timer_name
        self._context = context
        self._callback = callback
        self._is_first_call = True
        self._ok = True
        self._task = asyncio.ensure_future(self._job())
        print(timer_name + " init done")

    async def _job(self):
        try:
            while self._ok:
                if not self._is_first_call or not self._first_immediately:
                    await asyncio.sleep(self._interval)
                await self._callback(self._name, self._context, self)
                self._is_first_call = False
        except Exception as ex:
            print(ex)

    def cancel(self):
        self._ok = False
        self._task.cancel()


async def some_callback(timer_name, context, timer):
    context['count'] += 1
    print('callback: ' + timer_name + ", count: " + str(context['count']))

    if timer_name == 'Timer 2' and context['count'] == 3:
        timer.cancel()
        print(timer_name + ": goodbye and thanks for all the fish")


timer1 = Timer(interval=1, first_immediately=True, timer_name="Timer 1", context={'count': 0}, callback=some_callback)
timer2 = Timer(interval=5, first_immediately=False, timer_name="Timer 2", context={'count': 0}, callback=some_callback)

try:
    loop = asyncio.get_event_loop()
    loop.run_forever()
except KeyboardInterrupt:
    timer1.cancel()
    timer2.cancel()
    print("clean up done")

The solution proposed by Mikhail has one drawback. Calling cancel() cancels both: the timer and the actual callback (if cancel() fired after timeout is passed, but actual job is still in progress). Canceling the job itself may be not the desired behavior.

An alternative approach is to use loop.call_later:

async def some_job():
    print('Job started')
    await asyncio.sleep(5)
    print('Job is done')

loop = asyncio.get_event_loop() # or asyncio.get_running_loop()

timeout = 5
timer = loop.call_later(timeout, lambda: asyncio.ensure_future(some_job()))

timer.cancel() # cancels the timer, but not the job, if it's already started