Можно использовать celery для выполнения периодических заданий. Для этого надо настроить celery (самый короткий и ясный гайд), и использовать "агента" celery - beat_schedule
, который будет периодически запускать задачи.
Например:
#celery_settings.py
import os
import celery
os.environ.setdefault("FASTAPI_SETTINGS_MODULE", "service/settings.env")
app = celery.Celery("fastapi_celery", broker="redis://localhost:6379")
app.config_from_object("settings.env", namespace="CELERY")
app.autodiscover_tasks()
#настраиваем beat_schedule
app.conf.beat_schedule = {
"parsing_steam":
{
"task": "tasks.table_update",
"schedule": 60.0,
},
"updating_ml":
{
"task": "tasks.update_weights",
"schedule": 3000.0,
}
}
Перед этим рекомендуется переместить функции-задачи celery в отдельную директорию, и использовать декоратор @*твой_app*.task
или создать новое определение с декоратором @*твой_app*.task
, в котором будет вызываться старая функция (всё это для чистоты кода).
Например:
from celery_settings import app
from data.parser.parser import table_update
from service.ml.ml import update_weights
@app.task
def table_update():
table_update()
@app.task
def update_weights():
update_weights()
Это должно работать. POV celery - широко используемая асинхронная очередь задач, которая используется вместе с различными API, от FastApi, Django до aiogram.