Better FastAPI Background Jobs

Posted: 2022-08-29

Better Background Tasks for FastAPI or Discord.py

The built-in background tasks from starlette are useful for running a job that you don’t want to make the API response wait for, but they lack some features I was looking for such as; repeated runs, point in time runs, and cron-like functionality

• Pros
  • Runs after response is returned
  • Simply runs jobs in the same runtime
• Cons
  • Didn’t support Cron or Future Scheduled jobs
  • No ability to persist jobs across restarts

Advanced Python Scheduler (APScheduler)

Eventually I found APScheduler, which seemed to offer a nice amount on functionality without the need to set up a separate runtime for the jobs

• Pros
  • Simply runs jobs in the same runtime
  • Support Cron or Future Scheduled jobs
  • Ability to persist jobs across restarts
• Cons
  • Doesn’t automatically wait until after response is returned (solved with workaround)

import logging
from datetime import datetime, timedelta

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI

scheduler = AsyncIOScheduler()

logging.basicConfig(
    format='%(asctime)s - %(process)s - %(name)s:%(lineno)d - %(levelname)s -'
    ' %(message)s',
    level=logging.INFO,
)
logger = logging.getLogger(__name__)


# setup preconfigured jobs to run using the decorator
@scheduler.scheduled_job('interval', minutes=1)
async def example_heartbeat():
    now = datetime.now()
    logger.info(f'Time: {now}')


app = FastAPI()


# Start the scheduler running with a fastapi startup job
@app.on_event('startup')
async def startup_jobs():
    scheduler.start()


async def my_job(registered_ts):
    now_ts = datetime.now()
    logger.info(f"Job {registered_ts=}, {now_ts=}")


@app.get('/')
async def root():
    now = datetime.now()
    when = now + timedelta(minutes=5)
    # schedule adhoc jobs as needed
    scheduler.add_job(
        my_job, 'date', run_date=when, kwargs={'registered_ts': now}
    )

Alternate implementation, not running until after response

from fastapi import BackgroundTasks


@app.get('/')
async def root(background_tasks: BackgroundTasks):
    now = datetime.now()
    when = now + timedelta(minutes=5)
    # schedule adhoc jobs as needed (but after response returns)
    background_tasks.add_task(
        scheduler.add_job, my_job, 'date', run_date=when, kwargs={'registered_ts': now}
    )

What this enabled me to do with my Home Automation projects

• Automatically switch of lights in my home
  • at particular times of the day
  • after a fixed delay since a motion sensor last activated or another trigger occurred
• Periodically check if conditions had been met to send me alerts

Other options I considered

Repeated Tasks from fastapi-utils

• Pros
  • Support for periodic jobs from startup (was sufficient for my initial use-cases)
  • Simply runs jobs in the same runtime
• Cons
  • Didn’t support Cron or Future Scheduled jobs
  • No ability to persist jobs across restarts

Celery

Seemed a bit too heavy weight, requiring more complex configuration