-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
40 lines (25 loc) · 789 Bytes
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import asyncio
import time
from async_schedule import schedule
# function
def foo():
print("foo")
# coroutine function
async def bar():
print("bar")
# enqueue function in scheduler
schedule.every().second.do(foo)
# enqueue coroutine function in scheduler
schedule.every(2).seconds.do(bar) # no call of coroutine function here, just add the coroutine function
# synchronous way shall work as expected from schedule module
while True:
schedule.run_pending()
time.sleep(1)
# run within running asyncio loop by enqueuing schedule.run_pending in the running loop
async def main():
loop = asyncio.get_running_loop()
while True:
loop.call_soon(schedule.run_pending)
await asyncio.sleep(1)
if __name__ == "__main__":
asyncio.run(main())