Python APScheduler - How does AsyncIOScheduler work?

Question

I'm having a hard time understanding how the AsyncIOScheduler works, and how is it non blocking?

If my job is executing a blocking function, will the AsyncIOScheduler be blocking?

And what if I use AsyncIOScheduler with ThreadPoolExecutor? How does that work? Can I await the job execution?

Answer 1

Using some Internet resources I found some useful facts. Hope it will help you.

A typical APScheduler instance houses tens of jobs, which execute regular Python functions. There is no limit on the number of jobs an APScheduler instance can schedule; it only depends on the actual load of the machine. By default, APScheduler stores all jobs in-memory. If you want your jobs to survive from process restarts and keep triggering from the last time there were triggered, you can store these jobs in a database, such as any RDBMS, Redis, MongoDB, etc.

Depending on how your applications run, it can run as a thread, or an asyncio task, or else. When initialized, APScheduler doesn't do anything unless you add the Python functions as jobs. Once all the jobs are added, you need to "start" the scheduler. For a simple example of how to use APScheduler, here is a snippet of code that just works.

from urllib.request import urlopen
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

@scheduler.scheduled_job("interval", seconds=10)
def keep_warm():
    urlopen("https://enqueuezero.com", timeout=10)
    
scheduler.start()

This makes sure a URL is requested every 10 seconds. The program runs as a blocking process. If you want to co-exist them with your application, you can consider using BackgroundScheduler, AsyncIOScheduler, etc.

Here some code snippets for BackgroundScheduler.

from datetime import datetime
import time
import os

from apscheduler.schedulers.background import BackgroundScheduler


def tick():
    print('Tick! The time is: %s' % datetime.now())


if __name__ == '__main__':
    scheduler = BackgroundScheduler()
    scheduler.add_job(tick, 'interval', seconds=3)
    scheduler.start()
    print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

    try:
        # This is here to simulate application activity (which keeps the main thread alive).
        while True:
            time.sleep(2)
    except (KeyboardInterrupt, SystemExit):
        # Not strictly necessary if daemonic mode is enabled but should be done if possible
        scheduler.shutdown()

The below code will demonstrate how to use the asyncio compatible scheduler to schedule a job that executes on 3-second intervals.

import asyncio
import os
from datetime import datetime

from apscheduler.schedulers.asyncio import AsyncIOScheduler


def tick():
    print('Tick! The time is: %s' % datetime.now())


if __name__ == '__main__':
    scheduler = AsyncIOScheduler()
    scheduler.add_job(tick, 'interval', seconds=3)
    scheduler.start()
    print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

    # Execution will block here until Ctrl+C (Ctrl+Break on Windows) is pressed.
    try:
        asyncio.get_event_loop().run_forever()
    except (KeyboardInterrupt, SystemExit):
        pass

You can get more examples about apscheduler using this link!

Answer 2

Based on the documentation an AsyncIOScheduler is executed in an event loop.

It is non-blocking because it will simply add itself to the event loop and wait until you start it.

Once the event loop is started it will run asynchronously.

from apscheduler.schedulers.asyncio import AsyncIOScheduler
import asyncio

async def job():
    print('hi')

scheduler = AsyncIOScheduler()
scheduler.add_job(job, "interval", seconds=3)

scheduler.start()

asyncio.get_event_loop().run_forever()

Output

Run time of job "job (trigger: interval[0:00:03], next run at: 2020-07-27 14:06:39 -03)" was missed by 0:00:02.542515
hi
hi