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())