Scheduling in Python

Scheduling tasks is an essential part of any web application, especially those that require periodic or delayed actions. There are many ways to schedule tasks in Python, each with strengths and weaknesses. In this article, we'll look at some of the most popular ways to schedule tasks in a FastAPI app.

sched - Event scheduler from Python

The sched module is part of Python's standard library and provides a simple way to schedule events in a program. While it can be used in a FastAPI app, it is not recommended due to its simplicity and limited functionality.

Here's an example of using sched to schedule a task in a FastAPI app:

import sched
import time

scheduler = sched.scheduler(time.time, time.sleep)

def run_me_every_minute():
    print("Running every minute...")

def schedule_next_event():
    scheduler.enter(60, 1, run_me_every_minute)
    scheduler.run()

scheduler.enter(60, 1, run_me_every_minute)
while True:
    schedule_next_event()

In the above code, we create a scheduler object using the sched module and define a function run_me_every_minute to run as a scheduled task. We then use the enter method to schedule the task to run after 60 seconds and the run method to start the scheduler.

While sched is simple and easy to use, but lacks some of the advanced features of other scheduling libraries.

schedule python package

The schedule package (-> Github, -> Docs) describes itself as "Python job scheduling for humans." It provides a  powerful and flexible way to schedule tasks in Python. It is easy to use and has a simple API.

Here's an example of using schedule to schedule a task in a FastAPI app:

import schedule
import time

def run_me_every_minute():
    print("Running every minute...")

schedule.every(1).minutes.do(run_me_every_minute)

while True:
    schedule.run_pending()
    time.sleep(1)

In the above code, we define a function run_me_every_minute to run as a scheduled task and use the every method to schedule the task to run every minute. We then use the run_pending method to check for scheduled tasks and the sleep method to delay the loop for 1 second.

While schedule is easy to use and provides many advanced features such as error handling and interval customization; it is limited to running on a single thread and may not be suitable for high-performance applications.

Using repeated tasks from fastapi-utils

The fastapi-utils package (-> Github, -> Docs) provides a simple and convenient way to schedule repeated tasks in a FastAPI app. It uses the asyncio library to handle asynchronous tasks.

Here's an example of using fastapi-utils to schedule a task in a FastAPI app:

from fastapi import FastAPI
from fastapi_utils.tasks import repeat_every

app = FastAPI()

@repeat_every(seconds=60)
async def run_me_every_minute():
    print("Running every minute...")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app)

In the above code, we use the repeat_every decorator to schedule a task to run every 60 seconds. We then define a function run_me_every_minute to run as the scheduled task.

fastapi-utils is easy to use and integrates well with FastAPI, but it may not be suitable for more complex scheduling tasks.

Using package arq

The arq package (-> Github, -> Docs) provides a powerful and flexible way to schedule tasks in a FastAPI app. It is built on top of the asyncio library and supports advanced features such as task priorities and retry strategies.

Here's an example of using arq to schedule a task in a FastAPI app:

from fastapi import FastAPI
from arq import create_pool
from arq.jobs import Job

app = FastAPI()

async def run_me_every_minute():
    print("Running every minute...")

@app.on_event("startup")
async def startup():
    redis_settings = {"host": "localhost", "port": 6379}
    pool = await create_pool(redis_settings)
    job = Job(run_me_every_minute)
    await pool.enqueue_job(job)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app)

In the above code, we use the create_pool function to create a connection pool to Redis, which is used by arq to store and schedule tasks. We then define a function run_me_every_minuteto run as the scheduled task and create a Job object to encapsulate the task. Finally, we use the enqueue_job method to schedule the task to run.

arq is a powerful and flexible scheduling library that can handle complex scheduling tasks, but it requires more setup and configuration than some of the other libraries.

Using celery

Celery is a popular distributed task queue that can be used to schedule tasks in a FastAPI app. It supports various advanced features such as task priorities, retry strategies and task result storage. You find more about the Periodic tasks in the Celery documentation.

Here's an example of using Celery to schedule a task in a FastAPI app:

from fastapi import FastAPI
from celery import Celery

app = FastAPI()
celery = Celery('tasks', broker='pyamqp://guest@localhost//')

@celery.task
def run_me_every_minute():
    print("Running every minute...")

@app.on_event("startup")
async def startup():
    celery.conf.beat_schedule = {
        "run-every-minute": {
            "task": "main.run_me_every_minute",
            "schedule": crontab(minute="*")
        }
    }
    celery.conf.timezone = "UTC"
    celery.conf.task_routes = {"main.run_me_every_minute": {"queue": "default"}}
    celery.autodiscover_tasks(["main"])
    # Other Celery config


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app)

In the above code, we use the celery.schedules.crontab method to schedule the run_me_every_minute task to run every minute. We also define additional Celery configuration settings for timezone and task routing.

Celery is a powerful and feature-rich scheduling library that can handle complex scheduling tasks, but it requires more setup and configuration than some of the other libraries.

Using Dramatiq

Dramatiq is a high-performance distributed task-processing library that can schedule tasks in a FastAPI app. It supports advanced features such as task priorities, retry strategies, and result storage.

Here's an example of using Dramatiq to schedule a task in a FastAPI app:

from fastapi import FastAPI
from dramatiq import pipeline, actor, run_pipeline, cron

app = FastAPI()

@actor(cron("*/1 * * * *"))
def run_me_every_minute():
    print("Running every minute...")

@app.on_event("startup")
async def startup():
    job = pipeline(run_me_every_minute.message())
    run_pipeline(job)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app)

n the above code, we use the dramatiq.cron method to schedule the run_me_every_minute task to run every minute. The cron schedule is defined using the */1 * * * * expression, which means "every minute". We then use the pipeline and run_pipeline methods to schedule and execute the task.

Dramatiq is a powerful and high-performance scheduling library that can handle complex scheduling tasks, but it requires more setup and configuration than other libraries.

APScheduler

The APScheduler library is a popular and powerful scheduling library for Python. It provides many advanced features such as cron-style scheduling, interval scheduling, and more.

Here's an example of using APScheduler to schedule a task to run every minute in a FastAPI app:

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

app = FastAPI()

scheduler = AsyncIOScheduler()

async def run_me_every_minute():
    print("Running every minute...")

@app.on_event("startup")
async def startup():
    scheduler.add_job(run_me_every_minute, "interval", minutes=1)
    scheduler.start()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app)

In the above code, we use the APScheduler library to schedule a task run_me_every_minute to run every minute. We define a scheduler object and use the add_job method to add the task to the scheduler. The start method is then called to start the scheduler.

Note that we use the AsyncIOScheduler class to create an asynchronous scheduler. This is necessary when using APScheduler with FastAPI, as it is an asynchronous web framework.

In summary, APScheduler is a powerful scheduling library that provides many advanced features. It is compatible with FastAPI and can be used to schedule tasks to run at various intervals.

Conclusion

There are many ways to schedule tasks in a FastAPI app using Python, each with strengths and weaknesses. Choosing the right scheduling library depends on the complexity of your application and the specific features you need.

If you need a simple and easy-to-use scheduling library, the schedule package or fastapi-utils may be a good choice. If you need more advanced features and greater flexibility, arq, Celery, or Dramatiq may be more suitable.

No matter which library you choose, scheduling tasks can greatly enhance the functionality and efficiency of your FastAPI app. With the code examples provided in this article, you should have a good starting point for implementing task scheduling in your own projects.