Сообщить об ошибке.

Расширение Flask-APScheduler

Улучшенный планировщик для приложений Flask

Модуль Flask-APScheduler - это расширение Flask, добавляющее поддержку модуля APScheduler (улучшенный планировщик Python).

Основная функциональность модуля Flask-APScheduler:

  • Загружает конфигурацию планировщика из конфигурации Flask.
  • Загружает определения заданий из конфигурации Flask.
  • Позволяет указать имя хоста, на котором будет работать планировщик.
  • Предоставляет REST API для управления запланированными заданиями.
  • Обеспечивает аутентификацию для REST API.
  • Интегрируется с Flask Blueprints

Содержание:

Установка модуля Flask-APScheduler в виртуальное окружение.

Так как модуль Flask-APScheduler не входит в стандартную библиотеку Python, его необходимо установить отдельно. сделать это можно с помощью менеджера пакетов pip.

# создаем виртуальное окружение, если нет
$ python3 -m venv .venv --prompt VirtualEnv
# активируем виртуальное окружение 
$ source .venv/bin/activate
# обновляем `pip`
(VirtualEnv):~$ python3 -m pip install -U pip
# ставим модуль `Flask-APScheduler`
(VirtualEnv):~$ python3 -m pip install Flask-APScheduler

Параметры конфигурации Flask-APScheduler.

Параметры конфигурации, специфичные для Flask-APScheduler:

  • SCHEDULER_API_ENABLED: bool, значение по умолчанию: False - включает/отключает API планировщика;
  • SCHEDULER_API_PREFIX: str, значение по умолчанию: "/scheduler" - URL-префикс API планировщика;
  • SCHEDULER_ENDPOINT_PREFIX: str, значение по умолчанию: "scheduler." префикс конечной точки планировщика;
  • SCHEDULER_ALLOWED_HOSTS: list, значение по умолчанию: ["*"] (разрешено везде) - список разрешенных хостов, на которых будет запускаться планировщик (удобно при отладке планировщика разрешить выполнение заданий только на ['localhost', '127.0.0.1']).

Параметры конфигурации, характерные для APScheduler:

  • SCHEDULER_JOBSTORES: dict;
  • SCHEDULER_EXECUTORS: dict;
  • SCHEDULER_JOB_DEFAULTS: dict;
  • SCHEDULER_TIMEZONE: dict.

Для получения более подробной информации ознакомьтесь с документацией модуля APScheduler.

Базовое использования планировщика в приложении Flask.

  • Создаем приложение Flask.
  • Импортируем и инициализируем расширение Flask-APScheduler.
  • Устанавливаем любую необходимую конфигурацию.
from flask import Flask
from flask_apscheduler import APScheduler

# конфигурация приложения Flask
class Config:
    SCHEDULER_API_ENABLED = True

# инициализируем планировщик
scheduler = APScheduler()
# добавляем задание планировщика
@scheduler.task("interval", id="do_job_1", seconds=30, misfire_grace_time=900)
def job1():
    print("Job 1 выполняется")

if __name__ == '__main__':
    # создаем приложение Flask
    app = Flask(__name__)
    # читаем/применяем параметры 
    # конфигурации `Config` в приложение Flask
    app.config.from_object(Config())
    # если не хотите задавать параметры планировщика 
    # в конфигурации приложения (класс `Config`), 
    # то их можно установить здесь:
    # scheduler.api_enabled = True
    scheduler.init_app(app)
    # запускаем планировщик    
    scheduler.start()
    app.run()

Добавление заданий в планировщик Flask.

Задания могут быть добавлены в планировщик перед его запуском. Они создаются в оформленных функциях, которые должны быть импортированы перед вызовом app.run().

# инициализируем планировщик
scheduler = APScheduler()

# включаем планировщик
scheduler.api_enabled = True

# пример выполнения задания через интервал 'interval'
@scheduler.task('interval', id='do_job_1', seconds=30, misfire_grace_time=900)
def job1():
    print('Job 1 выполняется')

# примеры с 'cron'
@scheduler.task('cron', id='do_job_2', minute='*')
def job2():
    print('Job 2 выполняется')

@scheduler.task('cron', id='do_job_3', week='*', day_of_week='sun')
def job3():
    print('Job 3 выполняется')

# запускаем планировщик
scheduler.start()

Задания также можно добавлять после запуска планировщика.

scheduler.start()
scheduler.add_job(<id>,<function>, **kwargs)

API планировщика Flask-APScheduler.

Следующие команды могут быть переданы планировщику и довольно понятны сами по себе:

  • scheduler.start() - запускает планировщик приложения Flask;
  • scheduler.shutdown() - завершает работу планировщика;
  • scheduler.pause() - останавливает запуск любых заданий, уже запущенные задания не затронуты;
  • `scheduler.resume() > начинает выполнение запланированных заданий после приостановки;
  • scheduler.add_listener(<callback function>,<event>) - добавляет
  • scheduler.remove_listener(<callback function>) - удаляет
  • scheduler.add_job(<id>,<function>, **kwargs) - добавляет задание в планировщик;
  • scheduler.remove_job(<id>, **<jobstore>) - удаляет задание по его индификатору из хранилища заданий;
  • scheduler.remove_all_jobs(**<jobstore>) - удаляет все задания;
  • scheduler.get_job(<id>,**<jobstore>) - возвращает сведений о задании;
  • scheduler.modify_job(<id>,**<jobstore>, **kwargs) - обновляет уже существующее задание;
  • scheduler.pause_job(<id>, **<jobstore>) - приостанавливает задание;
  • scheduler.resume_job(<id>, **<jobstore>) - возобновляет задание;
  • scheduler.run_job(<id>, **<jobstore>) - немедленно запускает задание;

REST API планировщика Flask-APScheduler

Модуль Flask-APScheduler поставляется со встроенным API. Это API можно включить/отключить в конфигурации приложения Flask параметром SCHEDULER_API_ENABLED: True.

  • /scheduler [GET]: возвращает основную информацию о веб-планировщике;
  • /scheduler/pause [POST]: приостанавливает обработку задания в планировщике;
  • /scheduler/resume [POST]: возобновляет обработку задания в планировщике;
  • /scheduler/start [POST]: запускает планировщик;
  • /scheduler/shutdown [POST]: завершает работу планировщика с помощью wait=True;
  • /scheduler/shutdown [POST] с данными json={‘wait’:False}: завершает работу планировщика с помощью wait=False;
  • /scheduler/jobs [POST json job data]: добавляет задание в планировщик;
  • /scheduler/jobs/<job_id> [GET]: возвращает json-файл сведений о задании;
  • /scheduler/jobs [GET]: возвращает json с подробной информацией обо всех заданиях;
  • /scheduler/jobs/<job_id> [DELETE]: удаляет задание из планировщика;
  • /scheduler/jobs/<job_id> [PATCH json job data]: обновляет уже существующее задание;
  • /scheduler/jobs/<job_id>/pause [POST]: приостанавливает задание, возвращает сведений о задании в json;
  • /scheduler/jobs/<job_id>/resume [POST]: возобновляет задание, возвращает сведений о задании в json;
  • /scheduler/jobs/<job_id>/run [POST]: немедленно запускает задание, возвращает json с подробностями задания.

Пример описания заданий в конфигурации приложения.

from flask import Flask
from flask_apscheduler import APScheduler

# конфигурация приложения Flask.
class Config:
    # описание задания в конфигурации приложения
    JOBS = [
        {
            "id": "job1",
            "func": "jobs:job1",
            "args": (1, 2),
            "trigger": "interval",
            "seconds": 10,
        }
    ]
    # включаем API планировщика 
    SCHEDULER_API_ENABLED = True

# функция-задание планировщика
def job1(var_one, var_two):
    print(f'{str(var_one)} : {str(var_two)}')

if __name__ == "__main__":
    app = Flask(__name__)
    # читаем конфигурацию
    app.config.from_object(Config())
    # инициализируем планировщик
    scheduler = APScheduler()
    # добавляем планировщик в приложение Flask
    scheduler.init_app(app)
    # запускаем планировщик
    scheduler.start()
    app.run()

Пример указания разрешенных доменов для выполнения заданий планировщика.

from flask import Flask
from flask_apscheduler import APScheduler

# конфигурация приложения Flask.
class Config:
    # описание задания в конфигурации приложения
    JOBS = [
        {
            "id": "job1",
            # разрешить выполнения задания `job1` 
            # только на разрешенных хостах
            "func": "allowed_host:job1",
            "args": (1, 2),
            "trigger": "interval",
            "seconds": 10,
        }
    ]
    # указываем разрешенные хосты 
    SCHEDULER_ALLOWED_HOSTS = ["127.0.0.1", "localhost"]
    # включаем API планировщика 
    SCHEDULER_API_ENABLED = True

# функция-задание планировщика
def job1(var_one, var_two):
    print(f'{str(var_one)} : {str(var_two)}')

if __name__ == "__main__":
    app = Flask(__name__)
    # читаем конфигурацию
    app.config.from_object(Config())

    scheduler = APScheduler()
    # также можно задать список разрешенных серверов напрямую 
    # scheduler.allowed_hosts = ['my_servers_name'] 
    scheduler.init_app(app)
    scheduler.start()
    app.run()

Пример использования расширенных параметров конфигурации планировщика.

Рассмотрен пример хранения заданий в базе данных.

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from flask import Flask
from flask_apscheduler import APScheduler

# конфигурация приложения Flask.
class Config:
    JOBS = [
        {
            "id": "job1",
            "func": "advanced:job1",
            "args": (1, 2),
            "trigger": "interval",
            "seconds": 10,
        }
    ]
    SCHEDULER_JOBSTORES = {"default": SQLAlchemyJobStore(url="sqlite://")}
    SCHEDULER_EXECUTORS = {"default": {"type": "threadpool", "max_workers": 20}}
    SCHEDULER_JOB_DEFAULTS = {"coalesce": False, "max_instances": 3}
    SCHEDULER_API_ENABLED = True

# функция-задание планировщика
def job1(var_one, var_two):
    print(str(var_one) + " " + str(var_two))


if __name__ == "__main__":
    app = Flask(__name__)
    app.config.from_object(Config())
    scheduler = APScheduler()
    scheduler.init_app(app)
    scheduler.start()
    app.run()