How to implement scheduled tasks and periodic tasks in FastAPI

How to implement scheduled tasks and periodic tasks in FastAPI

Introduction:
FastAPI is a modern, highly performant Python framework focused on building API applications. However, sometimes we need to perform scheduled tasks and periodic tasks in FastAPI applications. This article describes how to implement these tasks in a FastAPI application and provides corresponding code examples.

1. Implementation of scheduled tasks

1. Using APScheduler library
   APScheduler is a powerful Python library for scheduling and managing scheduled tasks. It supports multiple task schedulers, such as based on date, time interval and Cron expression. The following are the steps to use APScheduler to implement scheduled tasks in FastAPI:

   1. Install the APScheduler library: Run the command pip install apscheduler in the terminal to install the APScheduler library.
   2. Create a scheduled task module: In the root directory of the FastAPI application, create a file named tasks.py to define scheduled tasks.

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

@scheduler.scheduled_job('interval', seconds=10)
def job():
    print("This is a scheduled job")

scheduler.start()

3. Register the scheduled task module: In the main file of the FastAPI application, import the scheduled task module and register it as a sub-application of the FastAPI application.

from fastapi import FastAPI
from .tasks import scheduler

app = FastAPI()

app.mount("/tasks", scheduler.app)

2. Using Celery library
   Celery is a powerful distributed task queue library that supports asynchronous and scheduled tasks. The following are the steps to use Celery to implement scheduled tasks in FastAPI:

   1. Install the Celery library: Run the command pip install celery in the terminal to install the Celery library.
   2. Create a scheduled task module: In the root directory of the FastAPI application, create a file named tasks.py to define scheduled tasks.

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379')

@app.task
def job():
    print("This is a scheduled job")

3. Register the scheduled task module: In the main file of the FastAPI application, import the scheduled task module and register it as a sub-application of the FastAPI application.

from fastapi import FastAPI
from .tasks import app as celery_app

app = FastAPI()

app.mount("/tasks", celery_app)

2. Implementation of periodic tasks

1. Using the APScheduler library
   The APScheduler library also supports the scheduling of periodic tasks. The following are the steps to use APScheduler to implement periodic tasks in a FastAPI application:

   1. Install the APScheduler library: Refer to step 1 in the previous article.
   2. Create a periodic task module: refer to step 2 in the previous article.

from apscheduler.triggers.cron import CronTrigger

scheduler = BackgroundScheduler()

@scheduler.scheduled_job(CronTrigger.from_crontab('* * * * *'))
def job():
    print("This is a periodic job")

scheduler.start()

2. Using the Celery library
   The Celery library also supports the scheduling of periodic tasks. The following are the steps to use Celery to implement periodic tasks in a FastAPI application:

   1. Install the Celery library: Refer to step 1 in the previous article.
   2. Create a periodic task module: refer to step 2 in the previous article.

from celery import Celery
from celery.schedules import crontab

app = Celery('tasks', broker='redis://localhost:6379')

@app.task
def job():
    print("This is a periodic job")

app.conf.beat_schedule = {
    'job': {
        'task': 'tasks.job',
        'schedule': crontab(minute='*'),
    },
}

Conclusion:
By using APScheduler or Celery library, we can easily implement scheduled tasks and periodic tasks in FastAPI applications. The code examples provided above can be used as a reference to help you quickly implement these task functions in your FastAPI project. Although the above are simple examples, you can further extend and customize your own task logic based on these examples.

Reference materials:

• APScheduler official documentation: https://apscheduler.readthedocs.io/
• Celery official documentation: https://docs.celeryproject. org/

(This article is for reference only, please adjust and modify it accordingly according to actual needs.)

The above is the detailed content of How to implement scheduled tasks and periodic tasks in FastAPI. For more information, please follow other related articles on the PHP Chinese website!