С Python всегда есть много библиотек и вариантов решения какой -либо конкретной проблемы, и запуск запланированных или повторяющихся заданий не является исключением. Независимо от того, хотите ли вы выполнить простую отложенную задачу, кучу запланированных заданий или управлять вкладками Cron, в Python есть специализированная библиотека. Итак, в этой статье я дам вам обзор всех доступных вариантов, которые помогут вам выбрать правильный инструмент для поставленной задачи, а также их варианты использования, включая вступление и основные примеры, чтобы вы начали быстро!
Встроенное решение
Прежде чем изучить какие -либо внешние библиотеки, давайте сначала проверим, что у нас есть в стандартной библиотеке Pythons. В большинстве случаев библиотека Python Standard будет содержать решение любой проблемы, которую вы можете столкнуться, и если проблема выполняет отложенные задания, как в Linux в Команда, затем захват SHAD Модуль может быть способом.
SHAD является очень простым модулем, который можно использовать для планирования одноразовых задач в течение некоторого указанного времени – поэтому важно понять, что это не повторяющаяся работа (например, Он работает на всех платформах, что может показаться очевидным, но не обязательно будет иметь место со всеми библиотеками, показанными позже.
Один из вариантов использования для такой отложенной задачи может быть запланирован на выключение или, если вы работаете с сетевыми подключениями или брандмауэром, вы можете создать одноразовую работу, чтобы вернуть изменения в случае, если вы испортите и заблокируете себя из системы.
Достаточно разговоров, давайте посмотрим пример:
import sched
import threading
import time
scheduler = sched.scheduler(time.time, time.sleep)
def some_deferred_task(name):
print('Event time:', time.time(), name)
print('Start:', time.time())
now = time.time()
# delay in seconds -----v v----- priority
event_1_id = scheduler.enter(2, 2, some_deferred_task, ('first',))
event_2_id = scheduler.enter(2, 1, some_deferred_task, ('second',)) # If first 2 events run at the exact same time, then "second" is ran first
event_3_id = scheduler.enter(5, 1, some_deferred_task, ('third',))
# Start a thread to run the events
t = threading.Thread(target=scheduler.run)
t.start()
# Event has to be canceled in main thread
scheduler.cancel(event_2_id)
# Terminate the thread when tasks finish
t.join()
# Output:
# Start: 1604045721.7161775
# Event time: 1604045723.718353 first
# Event time: 1604045726.7194896 third
Приведенный выше код определяет планировщик, который используется для создания ( .Enter ), которые будут выполняться в более позднее время. Каждое событие (Call to .Enter ) получает 4 аргумента, которые являются – задержка в секундах ( за сколько секунд произойдет событие? аргументы. Приоритет Аргумент не имеет значения большую часть времени, но может быть очень важным, если планируется произойти 2 события в точно В то же время, но они должны быть выполнены последовательно. В этом случае событие с наивысшим приоритетом (наименьшее число) идет первым.
В этом фрагменте кода мы также можем увидеть, что .Enter Метод возвращает идентификатор события. Эти идентификаторы могут быть использованы для отмены событий, как показано с scheduler.cancel (event_2_id) .
Чтобы не блокировать основной поток программы, мы также использовали резьба. Тема Чтобы начать планировщик и называется .присоединиться () на нем изящно прекратить, когда это делается со всеми задачами.
Полная сила Crontab
Есть довольно много библиотек для выполнения повторяющихся заданий с помощью Python, но давайте начнем с того, что дает вам полный Cron “Опыт” Анкет Эта библиотека называется Python-Crontab и может быть установлен с PIP установить Python-Crontab Анкет
Python-Crontab , в отличие от других библиотек и модулей, перечисленных здесь, создает и управляет реальными реальными кронтабами в системах UNIX и задачах в Windows. Следовательно, это не подражает поведению этих инструментов операционной системы, а скорее использует их и использует то, что уже есть.
Для примера здесь давайте посмотрим на какой -то практическое использование. Общей причиной выполнения повторяющихся задач может быть проверка состояния сервера баз данных. Это может быть обычно сделано, подключившись к базе данных и запустив фиктивные запросы, такие как Выберите 1 , точно так же:
from crontab import CronTab
# user=True denotes the current user
cron = CronTab(user=True)
job = cron.new(command='PGPASSWORD=test psql -U someuser -d somedb -c "SELECT 1" -h localhost')
job.setall("*/5 * * * *")
if cron[0].is_valid(): # If syntax is valid, write to crontab
cron.write()
# crontab -l # Check real crontab from shell
# */5 * * * * PGPASSWORD=test psql -U someuser -d somedb -c "SELECT 1" -h localhost
Как я упоминал ранее, Python-Crontab обеспечивает настоящий крон “Опыт” , который включает в себя в целом неприязненный синтаксис Cron. Чтобы установить график, один использует .setall Метод для установки всех полей. Перед установкой графика, однако, нам нужно создать Crontab, используя Crontab () и укажите владельца пользователя. Если Верно передается, идентификатор, выполняющий пользователь, будет использоваться программой. Мы также должны создать индивидуальную работу ( .new () ) в этом кронтабе, проходящем в Команда быть выполненным и, необязательно, также Комментарий Анкет
Когда у нас есть Crontab и его работа, нам нужно написать его, но это хорошая идея проверить его синтаксис, используя .является действительным () Прежде чем мы это сделаем.
Другая основная задача администратора базы данных – это создание периодических резервных копий, которые также могут быть легко выполнены с помощью Python-Crontab , на этот раз с небольшим различным синтаксисом:
with CronTab(user='root') as cron: # with context manager cron.write() is called automatically
job = cron.new(
command='PGPASSWORD=test pg_dump -U someuser -d somedb -h localhost --column-inserts --data-only > backup.sql',
comment="Perform database backup"
)
job.every(2).days()
job.hour.on(1)
# job.every_reboot()
# job.hour.every(10)
# job.month.during('JAN', 'FEB') # Powerful but confusing/hard to parse syntax
# job.minute.during(15, 45).every(5)
# crontab -l
# 0 1 */2 * * PGPASSWORD=test pg_dump -U someuser -d somedb -h localhost --column-inserts --data-only > backup.sql # Perform database backup
Если вам не очень удобен синтаксис Cron, эта библиотека также предоставляет декларативный синтаксис, который показан в примере выше. Этот синтаксис, на мой взгляд, очень запутанный и даже сложнее читать и использовать, чем обычный синтаксис Cron, поэтому я бы рекомендовал придерживаться синтаксиса Cron или выбрать различную библиотеку (см. Следующий раздел).
Помимо разных синтаксисов, мы также можем увидеть использование контекстного менеджера Python, что позволяет нам опустить .Write Метод показан ранее. Еще одна вещь, которую нужно помнить, это то, что если вы решите запустить рабочие места Cron как корень Пользователь (не рекомендуется), как показано выше, тогда вам придется запустить программу с Sudo Анкет
Эта библиотека также имеет другие полезные функции, помимо базового создания и управления Crontabs. Один из них перечисляет и осматривает как пользовательские, так и систему Crontabs, а также поиск на основе критериев, таких как команда или комментарий конкретной работы:
from crontabs import CronTabs
for cron in CronTabs(): # Get list of all user and system crontabs
if cron.user:
print(f'{cron.user} has following cron jobs:')
else:
print(f'{cron.filen} has following cron jobs:')
for job in cron.crons:
print(f' {job.command}')
# martin has following cron jobs:
# PGPASSWORD=test psql -U someuser -d somedb -c "SELECT 1" -h localhost
# PGPASSWORD=test pg_dump -U someuser -d somedb -h localhost --column-inserts --data-only > backup.sql
# /etc/cron.d/anacron has following cron jobs:
# [ -x /etc/init.d/anacron ] && if [ ! -d /run/systemd/system ]; then /usr/sbin/invoke-rc.d anacron start >/dev/null; fi
# /etc/cron.d/popularity-contest has following cron jobs:
# test -x /etc/cron.daily/popularity-contest && /etc/cron.daily/popularity-contest --crond
# ...
jobs = CronTabs().all.find_command('psql') # lookup for all jobs running specific command
for job in jobs:
print(job)
# */5 * * * * PGPASSWORD=test psql -U someuser -d somedb -c "SELECT 1" -h localhost
Как я упоминал в предыдущем разделе, не все библиотеки, показанные здесь, работают точно так же на всех платформах. Python-Crontab Работает на Linux и Windows, но поддерживаются только в Windows Crontabs (задачи Windows).
Если вы действительно ненавидите синтаксис Cron
Мы видели, как планировать работу с декларативным синтаксисом с Python-Crontab В предыдущем разделе, но это не было на самом деле читаемой или удобной для пользователя. Если вы ищете самую удобную, самую популярную библиотеку с очень простым интерфейсом, то Расписание библиотека для вас.
Расписание основан на статье Переосмысление крона который описывает некоторые из проблем и слабостей Крона, и эта библиотека делает хорошую работу по решению их.
Самая большая жалоба на Cron, безусловно, – это его товар и трудно писать синтаксис, так что давайте посмотрим, как Расписание адресат:
import schedule
def task():
return ...
def task_with_args(value):
return ...
schedule.every(2).hours.do(task)
schedule.every().sunday.at("01:00").do(task)
schedule.every().hour.at(":15").do(task)
schedule.every(15).to(30).seconds.do(task) # Randomly between every 15 to 30 seconds
schedule.every().minute.do(task_with_args, "some value")
# Grouping jobs with tags
schedule.every().day.at("09:00").do(task).tag('daily', 'morning')
schedule.every().day.at("18:30").do(task).tag('daily', 'evening')
schedule.every().hour.do(task).tag('hourly')
schedule.clear('daily-tasks')
# No explicit "month" schedule
# No explicit "range" (during 10-14h; from Jan to Mar) schedule
Первые 5 запланированных выше рабочих мест на самом деле не нуждаются в большом объяснении. Код очень читается в довольно эксплуатационном. Интерфейс содержит только несколько функций в течение нескольких дней ( .monday () ) и times ( использовать.
Помимо простого планирования, интерфейс также содержит .ярлык () Метод для группировки заданий по тегу. Это может быть полезно, например, для отмены целых групп заданий (с .clear () ).
Одним из недостатков наличия такого простого интерфейса является отсутствие явного месяц или Диапазон Планирование, например, Планирование рабочих мест в течение 10-14 часов или от Яна до Мар на самом деле невозможно.
Помимо повторяющихся рабочих мест, вы также можете использовать Расписание запустить одноразовые задачи и достичь того же эффекта, что и при SHAD , но с более приятным синтаксисом:
# Execute one-off (deferred job)
def deferred_job():
# Do stuff...
return schedule.CancelJob
schedule.every().day.at('01:00').do(deferred_job)
while True:
schedule.run_pending()
time.sleep(1)
# To run in background - https://github.com/mrhwick/schedule/blob/master/schedule/__init__.py#L63
Помимо отложенной работы, этот фрагмент кода также показывает, что нам нужно сохранить тему для работы. Это потому, что эта библиотека не создает фактического крон или в рабочие места Если вы не хотите блокировать основной поток вашей программы, как в примере выше, вы также можете запустить ее в фоновом режиме, как показано Здесь Анкет
Все функции Вам может понадобиться
Все ранее упомянутые инструменты имеют свои плюсы и минусы, некоторые конкретные функции и дизайн, которые делают их хорошими для некоторых конкретных вариантов использования. Если вам, однако, вам нужно выполнять как отсроченные, так и периодические задания, необходимо хранить задания в базе данных, нуждаются в настройках функций ведения журнала и т. Д., То, скорее всего, ни один из вышеупомянутых инструментов не будет сокращать его.
Наиболее богатая и мощная библиотека для планирования работ любого рода в Python определенно Apscheduler , что означает Advanced Python Scheduler Анкет
Он помещает все коробки, когда речь идет о функциях, упомянутых выше, и такие функции требуют обширной конфигурации, поэтому давайте посмотрим, как Apscheduler Имеет ли это:
jobstores = {
'mongo': MongoDBJobStore(), # MongoDBJobStore requires PyMongo installed - `pip install pymongo`
'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite') # SQLAlchemyJobStore requires SQLAlchemy installed, Recommended to use PostgreSQL
}
executors = {
'default': ThreadPoolExecutor(20),
'processpool': ProcessPoolExecutor(5)
}
job_defaults = {
'coalesce': False,
'max_instances': 3
}
scheduler = BackgroundScheduler(jobstores=jobstores,
executors=executors,
job_defaults=job_defaults,
timezone=utc,
daemon=True) # Without daemonic mode, the main thread would exit. You can also keep it alive with infinite loop.
Этот фрагмент кода показывает пример конфигурации, которую можно использовать для настройки магазинов заданий SQLite и MongoDB, в которых размещаются запланированные задания. В нем показана конфигурация исполнителей, которые обрабатывают работу заданий – здесь мы указываем размер наших пулов. Мы также указываем некоторые по умолчанию задания, такие как количество экземпляров работы, которые могут работать параллельно. Все конфигурации передаются в планировщик, который используется для управления заданиями.
def task():
return ...
# trigger -> can also be 'cron' or 'date'
# misfire_grace_time -> seconds after the designated runtime that the job is still allowed to be run
# max_instances -> max number of concurrent instances
scheduler.add_job(task, trigger='interval', seconds=5, misfire_grace_time=600, max_instances=5)
# 'interval' trigger can take any args from https://apscheduler.readthedocs.io/en/latest/modules/triggers/interval.html#module-apscheduler.triggers.interval
scheduler.add_job(task, trigger='cron', month='jan-apr', day_of_week='mon-fri', hour=15, minute=30, end_date='2021-01-30')
scheduler.add_job(task, CronTrigger.from_crontab('0 17 * * sat,sun'))
# 'cron' trigger can take any args from https://apscheduler.readthedocs.io/en/latest/modules/triggers/cron.html#module-apscheduler.triggers.cron
# Simulates 'at' - deferred jobs
scheduler.add_job(task, trigger='date', run_date=datetime(2020, 12, 24, 17, 30, 0))
# 'date' trigger can take any args from https://apscheduler.readthedocs.io/en/latest/modules/triggers/date.html#module-apscheduler.triggers.date
scheduler.print_jobs(jobstore="default")
scheduler.start()
# Pending jobs:
# task (trigger: interval[0:01:00], pending)
# task (trigger: cron[month='jan-apr', day_of_week='mon-fri', hour='15', minute='30'], pending)
# task (trigger: cron[month='*', day='*', day_of_week='sat,sun', hour='17', minute='0'], pending)
# task (trigger: date[2020-12-24 17:30:00 UTC], pending)
Далее идет создание нашей работы с использованием .add_job () метод Требуется немало аргументов, в первую очередь они функционируют. Далее находится тип триггера, который может быть интервалом, кроном или датой. Интервальные графики задания периодически работают в указанном интервале. Cron просто старый добрый планировщик, похожий на Cron, который допускает классические и ключевые аргументы, основанные на аргументах, основанные на аргументах. Наконец, Date Trigger Создайте задания на определенную дату и время.
Еще один важный аргумент для .add_job () это misfire_grace_time , который предоставляет Анакрон – Подобная функция, или, другими словами – в случае, если ваша работа не работает, потому что планировщик не работает, планировщик попытается запустить ее, когда она будет резервным копированием, если misfire_grace_time не был превышен.
Запланированные работы, как правило, раздражают отладки. Apscheduler пытается облегчить это с возможностью легко настраивать уровни ведения журнала, а также возможность добавлять слушателей в события планировщика – например, Когда работа выполняется или когда работа не удается. Вы можете увидеть такой вывод для прослушивателя и примера журнала.
# Catching scheduler events:
import logging
logging.basicConfig(level=logging.INFO)
def my_listener(event):
if event.exception:
logging.warning('Job failed...')
else:
logging.info('Job was executed...')
scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
# ...
# INFO:apscheduler.scheduler:Scheduler started
# INFO:apscheduler.executors.default:Running job "task (trigger: interval[0:00:05], next run at: 2020-10-30 09:51:11 UTC)" (scheduled at 2020-10-30 09:51:11.222311+00:00)
# INFO:apscheduler.executors.default:Job "task (trigger: interval[0:00:05], next run at: 2020-10-30 09:51:16 UTC)" executed successfully
# INFO:root:Job was executed...
Для пользователей Gevent
Последнее и, возможно, на самом деле наименьшее (желаемое решение) – это использовать Gevent. Не потому, что Gevent плох, а потому, что он не на самом деле не создан для выполнения запланированных задач. Если вы, однако, уже используете Gevent в своем приложении, может иметь смысл использовать его для планирования заданий.
Если вы не знакомы с Gevent, то Gevent – это библиотека параллелистики, основанная на Coroutines. Он использует Greenlets, чтобы обеспечить псевдоконкуренду для выполнения нескольких задач в потоке одной ОС. Для лучшего понимания давайте посмотрим на основной пример:
# Plain Gevent without scheduler
import gevent
from gevent import monkey
# patches stdlib (including socket and ssl modules) to cooperate with other greenlets
monkey.patch_all()
import requests
# Note that we're using HTTPS, so
# this demonstrates that SSL works.
urls = [
'https://www.google.com/',
'https://www.twitter.com/',
'https://www.python.org/'
]
def head_size(url):
print(f'Starting {url}')
data = requests.get(url).content
print(f'{url}: {len(data)}')
jobs = [gevent.spawn(head_size, _url) for _url in urls]
gevent.wait(jobs)
# Output:
# Starting https://www.google.com/
# Starting https://www.twitter.com/
# Starting https://www.python.org/
# https://www.google.com/: 14381
# https://www.python.org/: 50202
# https://www.twitter.com/: 42702
Этот пример показывает, как мы можем запросить несколько URL -адресов параллельно, используя Gevent и его Gevent.spawn . В приведенном выше выводе вы можете видеть, что все 3 созданные задания начались в то же (-ish) и возвращали данные позже.
Чтобы выполнить ту же задачу, но запланировано в будущем, мы можем сделать следующее:
# deferred job (one-off)
def schedule(delay, func, *args, **kw_args):
gevent.spawn_later(0, func, *args, **kw_args)
gevent.spawn_later(delay, schedule, delay, func, *args, **kw_args)
schedule(30, head_size, urls[0])
# periodic job
# this will drift...
def run_regularly(function, interval, *args, **kwargs):
while True:
gevent.sleep(interval)
function(*args, **kwargs)
run_regularly(head_size, 30, urls[0])
# Output:
# Starting https://www.google.com/
# https://www.google.com/: 14449
# 30 sec later...
# Starting https://www.google.com/
# https://www.google.com/: 14435
# ...
Выше мы можем увидеть оба примера для выполнения одноразовых заданий, так и периодических. Оба эти решения являются своего рода взломом/трюком и должны рассматриваться только в том случае, если вы уже используете Gevent в своем приложении. Также важно упомянуть, что выше Run_regularly Функция постепенно начнет дрейфовать, даже если мы учитываем время выполнения задач. Поэтому вы должны предпочтительно использовать Geventschedule Доступно в Apscheduler Вместо этого библиотека, так как это более надежное решение.
Вывод
Запуск отложенных или периодических заданий – очень общая задача, и вы не найдете ни одной библиотеки или инструмента, который делает все это идеально, но я надеюсь, что эта статья дала вам приличный обзор того, что доступно. Независимо от того, какой инструмент вы выберете для вашего конкретного варианта использования, важно помнить о более общих практиках, например, например: добавьте комментарии к вашим заданиям Cron для ясности, избегайте использования корень Пользователь (принцип наименьшей привилегии), не вкладывайте пароли в свои Crontabs и т. Д. Кроме того, если вы собираетесь использовать настоящие задания Cron, вы также можете использовать /etc/cron.daily , /etc/cron.weekly и /etc/cron.mothly Чтобы сохранить организованные и аккуратные. И последнее, но не менее важное, изучение Anacrontab Может также быть полезным, если вам нужно убедиться, что ваши задания будут работать, даже если машина на некоторое время останется в автономном режиме.
Оригинал: “https://dev.to/martinheinz/scheduling-all-kinds-of-recurring-jobs-with-python-l74”