Asyncio

To use asyncio with the scheduler library, replace the default threading Scheduler (scheduler.threading.scheduler.Scheduler) with the asyncio Scheduler (scheduler.asyncio.scheduler.Scheduler) variant. Both schedulers provide a nearly identical API - the only difference is the lack of prioritization and weighting support for the asyncio Scheduler.

Note

In contrast to the threading Scheduler it is necessary to instanciate the asyncio Scheduler within a coroutine.

The following example shows how to use the asyncio Scheduler with a simple coroutine.

import asyncio
import datetime as dt

from scheduler.asyncio import Scheduler


async def foo():
    print("foo")


async def main():
    schedule = Scheduler()

    schedule.once(dt.timedelta(seconds=5), foo)
    schedule.cyclic(dt.timedelta(minutes=10), foo)

    while True:
        await asyncio.sleep(1)


asyncio.run(main())

To initialize the Scheduler with a user defined event loop, use the loop keyword argument:

import asyncio
import datetime as dt

from scheduler.asyncio import Scheduler


async def foo():
    print("foo")


async def main():
    loop = asyncio.get_running_loop()
    schedule = Scheduler(loop=loop)

    schedule.once(dt.timedelta(seconds=5), foo)
    schedule.cyclic(dt.timedelta(minutes=10), foo)

    while True:
        await asyncio.sleep(1)

asyncio.run(main())