如何在FastAPI中實現(xiàn)定時任務(wù)和週期性任務(wù)

引言：
FastAPI是一個現(xiàn)代化的、高度效能的Python框架，專注於建立API應(yīng)用程式。然而，有時我們需要在FastAPI應(yīng)用程式中執(zhí)行定時任務(wù)和週期性任務(wù)。本文將介紹如何在FastAPI應(yīng)用程式中實作這些任務(wù)，並提供對應(yīng)的程式碼範(fàn)例。

一、定時任務(wù)的實作

1. 使用APScheduler函式庫
   APScheduler是一個功能強大的Python函式庫，用於排程和管理定時任務(wù)。它支援多種任務(wù)調(diào)度器，如基於日期、時間間隔和Cron表達式等。以下是在FastAPI中使用APScheduler實作定時任務(wù)的步驟：

   1. 安裝APScheduler函式庫：在終端機中執(zhí)行指令pip install apscheduler來安裝APScheduler函式庫。
   2. 建立一個定時任務(wù)模組：在FastAPI應(yīng)用程式的根目錄下，建立一個名為tasks.py的文件，用於定義定時任務(wù)。

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. 註冊定時任務(wù)模組：在FastAPI應(yīng)用程式的主檔案中，匯入定時任務(wù)模組並註冊為FastAPI應(yīng)用程式的子應(yīng)用程式。

from fastapi import FastAPI
from .tasks import scheduler

app = FastAPI()

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

2. 使用Celery庫
   Celery是一個強大的分散式任務(wù)佇列庫，支援非同步和定時任務(wù)。以下是在FastAPI中使用Celery實作定時任務(wù)的步驟：

   1. 安裝Celery函式庫：在終端機中執(zhí)行指令pip install celery來安裝Celery函式庫。
   2. 建立一個定時任務(wù)模組：在FastAPI應(yīng)用程式的根目錄下，建立一個名為tasks.py的文件，用於定義定時任務(wù)。

from celery import Celery

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

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

3. 註冊定時任務(wù)模組：在FastAPI應(yīng)用程式的主檔案中，匯入定時任務(wù)模組並註冊為FastAPI應(yīng)用程式的子應(yīng)用程式。

from fastapi import FastAPI
from .tasks import app as celery_app

app = FastAPI()

app.mount("/tasks", celery_app)

二、週期性任務(wù)的實作

1. 使用APScheduler函式庫
   APScheduler函式庫同樣支援週期性任務(wù)的排程。以下是在FastAPI應(yīng)用程式中使用APScheduler實作週期性任務(wù)的步驟：

   1. 安裝APScheduler函式庫：參考前文中的步驟1。
   2. 建立一個週期性任務(wù)模組：參考前文中的步驟2。

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. 使用Celery函式庫
   Celery函式庫同樣支援週期性任務(wù)的排程。以下是在FastAPI應(yīng)用程式中使用Celery實作週期性任務(wù)的步驟：

   1. 安裝Celery函式庫：參考前文中的步驟1。
   2. 建立一個週期性任務(wù)模組：參考前文中的步驟2。

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='*'),
    },
}

結(jié)論：
透過使用APScheduler或Celery函式庫，我們可以輕鬆地在FastAPI應(yīng)用程式中實現(xiàn)定時任務(wù)和週期性任務(wù)。以上提供的程式碼範(fàn)例可以作為參考，幫助您在FastAPI專案中快速實現(xiàn)這些任務(wù)功能。儘管以上是簡單的範(fàn)例，但基於這些範(fàn)例，您可以進一步擴展和自訂自己的任務(wù)邏輯。

參考資料：

• APScheduler官方文件：https://apscheduler.readthedocs.io/
• Celery官方文件：https://docs.celeryproject. org/

（本文僅供參考，請依實際需求進行相應(yīng)調(diào)整和修改。）

以上是如何在FastAPI中實現(xiàn)定時任務(wù)和週期性任務(wù)的詳細內(nèi)容。更多資訊請關(guān)注PHP中文網(wǎng)其他相關(guān)文章！