How to Integrate APScheduler with FastAPI for Reliable Async Task Scheduling

Introduction

Scheduled tasks are a common requirement; after using Celery with Django, I explored APScheduler and built a demo integrated into a project.

Task scheduling mainly includes adding/deleting jobs, pausing/resuming (not implemented), and viewing job status.

Implementation

Adding Scheduled Jobs

There are three ways to add jobs:

date: execute once at a fixed time.

interval: execute repeatedly at fixed intervals.

cron: most flexible, using a crontab expression.

In FastAPI's async environment, use AsyncIOScheduler as the scheduler.

By default, SQLite is used to persist jobs so they survive restarts.

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.triggers.cron import CronTrigger

Schedule = AsyncIOScheduler(
    jobstores={
        'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
    }
)
Schedule.start()

Unfinished Parts

Parameter validation is incomplete due to time constraints.

Job‑parameter wrappers split into three functions, many parameters unused.

Documentation mentions max worker limits and potential execution‑time limits for long‑running functions.

Other issues need further investigation.

Alternative Solutions

Other scheduling options include: arq – lightweight async job queue (few stars on GitHub). fastapi-utils – provides utilities for FastAPI. celery – heavyweight Python task queue; can be used just for its scheduling features.

Code Repository

Single‑file example: https://github.com/CoderCharm/fastapi-mysql-generator/blob/master/examples/demo_scheduler/main.py

Usage in project: https://github.com/CoderCharm/fastapi-mysql-generator/blob/master/{{cookiecutter.project_name}}/app/api/__init__.py#L230

References

APScheduler official documentation

FastAPI issues