Python APScheduler FastAPI Integration for Background Jobs

When a FastAPI application needs to run periodic work—cache refreshes, cleanup jobs, metric aggregation, or polling an external API—the request/response model does not cover it. A common approach is to run a scheduler inside the same process as the web application, and APScheduler is the library most Python projects reach for. This article covers the practical side of python apscheduler fastapi integration: where the scheduler lives, how it starts and stops with the application, how to run async jobs, and what changes when you deploy to production.

Creating the Scheduler and Registering Jobs

APScheduler provides several scheduler classes. For a FastAPI application, the two relevant ones are BackgroundScheduler and AsyncIOScheduler.

BackgroundScheduler runs jobs in a thread pool and works with plain synchronous functions. AsyncIOScheduler integrates with the running asyncio event loop and can run coroutines directly. If your jobs are async—which is common when they call httpx, aiohttp, or an async database driver—you want AsyncIOScheduler.

A minimal setup with a synchronous job looks like this:

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

def cleanup_temp_files():
    # remove files older than 24 hours
    ...

def register_jobs(scheduler):
    scheduler.add_job(
        cleanup_temp_files,
        trigger="interval",
        hours=24,
        id="temp_cleanup",
        replace_existing=True,
    )

The id and replace_existing arguments matter. When you re-register jobs on every startup, replace_existing=True prevents duplicate entries with the same id. Without it, adding a job with an existing id raises a ConflictingIdError.

Managing the Scheduler Lifecycle with FastAPI Startup and Shutdown

The scheduler must start after the event loop is running and stop before the application shuts down. The modern way to hook into FastAPI's lifecycle is the lifespan context manager.

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    register_jobs(scheduler)
    scheduler.start()
    yield
    scheduler.shutdown(wait=False)

app = FastAPI(lifespan=lifespan)

start() launches the scheduler's background task. shutdown(wait=False) tells the scheduler not to wait for currently running jobs to finish; pass wait=True if in-flight jobs must complete before the process exits. The right choice depends on whether an interrupted job can leave the system in a bad state.

The older @app.on_event("startup") and @app.on_event("shutdown") decorators still work, but they are deprecated in recent FastAPI versions. The lifespan approach is the recommended path.

Running Async Jobs on the Event Loop

When you switch to AsyncIOScheduler, jobs can be coroutine functions:

from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()

async def refresh_cache():
    async with httpx.AsyncClient() as client:
        data = await client.get("https://api.example.com/feed")
    # store the result in cache
    ...

The scheduler runs the coroutine on the same event loop that handles incoming requests. That means a blocking call inside an async job—for example, time.sleep or a synchronous database driver—blocks all request handling for the duration. If a job is CPU-bound or uses blocking I/O, either keep it on BackgroundScheduler with a thread pool, or wrap the blocking portion in asyncio.to_thread inside an async job.

APScheduler 3.x is the stable release line and the API shown above applies to it. APScheduler 4.x, currently in alpha, changes the API significantly: the scheduler class is AsyncScheduler, jobs are added with add_schedule, and the configuration model is different. Do not mix the two APIs; pin the version you target in your dependencies.

Avoiding Duplicate Schedulers During Development

Running uvicorn app:app --reload starts two processes: the reloader and the worker. The lifespan runs in both, so you get two schedulers and jobs fire twice.

The common workaround is to check an environment variable that only the worker process sets:

import os

if os.environ.get("RUN_MAIN") == "true":
    register_jobs(scheduler)
    scheduler.start()

This is a development-only problem. In production with multiple workers, the same issue appears in a different form: each worker process runs its own scheduler, so a job that should run once per day runs once per worker. APScheduler has no built-in distributed coordination; if you need exactly-once execution across processes, you need an external lock or a dedicated scheduler.

Configuring Job Stores and Executors for Production

By default, APScheduler keeps jobs in memory and runs them in a thread pool. If the process restarts, all jobs are lost and must be re-registered. For most FastAPI deployments this is acceptable because jobs are defined in code and re-registered on startup.

If you need job state to survive restarts—for example, to track missed runs or to store jobs that were added dynamically—use a persistent job store:

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor

jobstores = {
    "default": SQLAlchemyJobStore(url="sqlite:///jobs.sqlite")
}
executors = {
    "default": ThreadPoolExecutor(10)
}

scheduler = AsyncIOScheduler(jobstores=jobstores, executors=executors)

The SQLAlchemyJobStore persists job definitions and next run times. It does not make jobs run exactly once across multiple processes; that still requires a distributed lock.

Observability and Error Handling for Scheduled Jobs

A scheduled job that raises an exception does not crash the scheduler. The exception is logged and the job is marked as failed. To catch failures centrally, attach a listener:

from apscheduler.events import EVENT_JOB_ERROR

def on_job_error(event):
    logger.error("Job %s failed", event.job_id)

scheduler.add_listener(on_job_error, EVENT_JOB_ERROR)

For long-running jobs, set max_instances=1 on the job so a slow run does not overlap with the next scheduled run. misfire_grace_time controls how long after a missed fire time the job is still allowed to run. With the default of one second, a job that was missed while the process was down will not run late on startup; raise the grace time if you want missed runs to be executed.

When a Different Scheduling Approach Makes Sense

If you only need a single delayed task after a request, FastAPI's BackgroundTasks is lighter than APScheduler and does not require lifecycle management. If you need cron-like scheduling across multiple processes with guaranteed single execution, a dedicated scheduler such as Celery beat or an external service is a better fit. APScheduler is the right choice when you want in-process scheduling with a simple API and do not need distributed coordination.