Планирование задач — неотъемлемая часть работы любых веб-приложений, в особенности таких, в которых требуется совершать периодические или отложенные действия. В Python предусмотрено множество способов планирования задач, и у каждого есть свои сильные и слабые стороны. В этой статье будут рассмотрены некоторые наиболее популярные способы планирования задач в приложении, написанном на основе FastAPI.

sched – планировщик событий из Python


Модуль sched входит в состав стандартной библиотеки Python и обеспечивает простой механизм для планирования событий в программе. Этот модуль может работать в приложении на FastAPI, но пользоваться им не рекомендуется, так как он слишком прост, и функциональность его ограничена.

Вот как при помощи sched можно запланировать задачу в приложении на FastAPI:

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()

В вышеприведённом коде мы создаём объект scheduler при помощи модуля sched и определяем функцию run_me_every_minute так, чтобы она работала как запланированная задача. Затем применяем метод enter, чтобы он назначал задачу, и она начинала выполняться через 60 секунд, а при помощи метода run запускаем планировщик.

Конечно, sched прост и лёгок в использовании, но в нём не хватает некоторых продвинутых возможностей, имеющихся в других библиотеках для планирования задач.

Пакет schedule для python


Пакет schedule (см. Github, см. документация) позиционируется как «планировщик задач Python, сделанный для людей». Он предоставляет мощные и гибкие способы планировать задачи в Python. Им удобно пользоваться и у него простой API.

На следующем примере показано, как при помощи schedule запланировать задачу в приложении на FastAPI:

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)

В вышеприведённом коде мы определяем функцию run_me_every_minute так, чтобы она выполнялась как запланированная задача, а при помощи метода every задаём выполнение этой задачи раз в минуту. Далее при помощи метода run_pending мы проверяем, какие задачи назначены, а при помощи метода sleep устанавливаем задержку цикла на 1 секунду.

Притом, что schedule прост в использовании, он ещё и предоставляет многие продвинутые возможности — в частности, обеспечивает обработку ошибок и подбор интервалов. Тем не менее, он может работать только в однопоточном режиме и для высокопроизводительных приложений не подойдёт.

Использование повторяющихся задач из fastapi-utils


Пакет fastapi-utils (см. Github, см. документация) позволяет просто и с удобством назначать повторяющиеся задачи в приложении, написанном на FastAPI. Для обработки асинхронных задач в нём используется библиотека asyncio.

Ниже показано, как при помощи fastapi-utils назначить задачу в приложении на FastAPI:

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)

В вышеприведённом коде при помощи декоратора repeat_every мы назначаем выполнение задачи с периодичностью в 60 секунд. Далее мы определяем функцию run_me_every_minute так, чтобы она выполнялась как назначенная задача.

Пакет fastapi-utils прост в использовании и хорошо интегрируется с FastAPI, но, возможно, не подойдёт для сравнительно сложных задач, связанных с планированием.

Использование пакета arq


Пакет arq (см. Github, см. документация) — ещё один мощный и гибкий инструмент для назначения задач в приложениях на FastAPI. Он выстроен на основе библиотеки asyncio и поддерживает продвинутые возможности: в частности, позволяет присваивать задачам приоритет и формулировать стратегии повторных попыток.

Вот как при помощи arq назначить задачу в приложении на FastAPI:

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)

В вышеприведённом коде мы при помощи функции create_pool создаём пул соединений с Redis, который arq затем использует для хранения и назначения задач. Мы определяем функцию run_me_every_minute так, чтобы она выполнялась как назначенная задача, а также инкапсулируем её в объект Job, который специально для этого создаём. Наконец, применяем метод enqueue_job, чтобы назначить эту задачу на выполнение.

arq — мощная и гибкая библиотека для планирования, справляющаяся со сложными задачами. Но с её настройкой и конфигурацией приходится поработать серьёзнее, чем в случае с некоторыми другими библиотеками.

Работа с celery


Celery — это популярный инструмент для работы с очередью распределённых задач, и его удобно применять при работе с приложением на FastAPI. В Celery поддерживаются различные продвинутые возможности, в частности, расстановка приоритетов задач, стратегии повторных попыток и хранилище для результатов задач. Подробнее о периодических задачах рассказано в документации по Celery.

Вот как при помощи Celery можно запланировать задачу в приложении на FastAPI:

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)

В вышеприведённом коде при помощи метода celery.schedules.crontab мы назначаем для Celery дополнительные конфигурационные настройки, так, чтобы задача run_me_every_minute выполнялась каждую минуту. Также мы определим для Celery конфигурационные настройки, в которых указывается часовой пояс и настройки маршрутизации задач.

Celery — это мощная и насыщенная возможностями библиотека для планирования задач, справляющаяся с очень сложными сценариями, но достаточно сложная в настройке и конфигурировании.

Работа с Dramatiq


Dramatiq — это высокопроизводительная библиотека для обработки распределённых задач, применимая с приложениями на FastAPI. В ней также поддерживаются такие возможности как расстановка приоритетов задач, стратегии повторных попыток и хранилище для результатов.

Вот как при помощи Dramatiq можно запланировать задачу в приложении на FastAPI:

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)

В вышеприведённом коде мы применяем метод dramatiq.cron, чтобы назначить ежеминутное выполнение задачи run_me_every_minute. Задача cron назначает для выполнения при помощи выражения */1 * * * *, означающего, что мы мы создаём конвейер, а затем задействуем методы pipeline и run_pipeline для назначения и выполнения задачи.

APScheduler


Библиотека APScheduler также очень мощная и пользуется популярностью при планировании задач на Python. В ней предусмотрено много продвинутых возможностей, в частности, планирование в стиле cron, интервальное планирование и многое другое.

Вот как при помощи APScheduler можно назначить, чтобы выбранная задача ежеминутно выполнялась в приложении на FastAPI:

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)

В вышеприведённом коде мы задаём при помощи библиотеки APScheduler, чтобы задача run_me_every_minute выполнялась ежеминутно. Мы определяем объект планировщика, а далее при помощи метода add_job добавляем задачу в планировщик. Далее для запуска планировщика вызывается метод start.

Обратите внимание: асинхронный планировщик создаётся при помощи класса AsyncIOScheduler. Он необходим при использовании APScheduler с FastAPI, поскольку FastAPI – это асинхронный фреймворк.

В целом APScheduler — это мощная библиотека для планирования, в которой предоставляется множество продвинутых возможностей. Она совместима с FastAPI и может использоваться при планировании задач, когда они должны выполняться с разными интервалами.

Заключение


При работе с приложениями, написанными на Python с применением FastAPI есть множество инструментов для планирования задач, каждый со своими достоинствами и недостатками. Подходящая библиотека для планирования выбирается в зависимости от сложности вашего приложения и от того, какие именно возможности вам требуются.

Если вам нужна простая и удобная в использовании библиотека для планирования, то вам хорошо подойдут варианты schedule или fastapi-utils. Если вам требуются более продвинутые возможности и более значительная гибкость при работе, попробуйте arq, Celery или Dramatiq.

Независимо от того, на какой библиотеке вы остановитесь, можно значительно усовершенствовать функциональность и эффективность вашего приложения на FastAPI, если настроить в нём планирование задач. Надеюсь, что приведённые в этой статье примеры кода станут для вас хорошей отправной точкой, чтобы реализовать планирование задач в ваших проектах.