Python: запуск скриптов через собственный планировщик либо через crontab
Если вы писали скрипты, которые должны выполняться с определенной периодичность, то наверное самое простое решение это повесить их выполнение на планировщик задач cron .
# открыть редактор планировщика > crontab -e # общий формат указания планировщику исполнение скрипта Python [параметры запуска] [ссылка на интерпретатор python] [путь до .py скрипта] # параметры запуска * * * * * - - - - - | | | | | | | | | ----- день недели (0—7) (воскресенье = 0 или 7) | | | ------- месяц (1—12) | | --------- день (1—31) | ----------- час (0—23) ------------- минута (0—59) # пример запуска "каждый час" 0 */1 * * * /usr/bin/python3 /opt/script1.py # пример запуска "каждые 4 часа в 15 минут" 15 */4 * * * /usr/bin/python3 /opt/script2.py # пример запуска "в 7:30 утра по воскресениям" 30 7 * * 0 /usr/bin/python3 /opt/script3.py
Пишем планировщик задач на Python с использованием библиотеки schedule
Есть и другой способ запускать функции python с определенной периодичностью без использования внешнего планировщика. Для этого будем использовать библиотеку schedule.
import datetime import schedule i = 1 def func(name): global i print(f'Функция запустилась раз') i += 1 # вывод времени чтобы проверить что все работает print(datetime.datetime.now()) # вызываем каждые 5 сек. schedule.every(5).seconds.do(func, name='schedule') # запускаем шедулер в бесконечном цикле для постоянного мониторинга факта наступления заданного события while True: schedule.run_pending()
В итоге исполнения данного кода, на выходе вы получите что-то похожее:
Функция schedule запустилась 1 раз 2022-04-11 11:18:34.124307 Функция schedule запустилась 2 раз 2022-04-11 11:18:39.124338 Функция schedule запустилась 3 раз 2022-04-11 11:18:44.124368 Функция schedule запустилась 4 раз 2022-04-11 11:18:49.124375
Как запускать метод каждую минуту?
Доброго времени суток. Подскажите, как мне запускать метод (функцию )в моей программе каждую минуту ?
Ответ зависит от целого ряда факторов.
1. Должна ли программа делать что-то ещё? Если да, то что именно?
2. Сколько таких периодических методов будет существовать? Строго один, или может быть более? Должны ли у них быть разные периоды?
Если программа синхронная, и метод только один, то можно использовать простой приём:
import time stop = False def proc_to_call(): global stop #Это твой периодический вызов #когда нужен останов, делаешь stop = True pass while not stop: time.sleep(60) proc_to_call() #код после цикла выполнится только после прерывания цикла.Если вызовов несколько и с разной периодичностью, то лучше использовать пакет schedule. Но у тебя всё равно будет бесконечный цикл вида
while True: schedule.run_pending() time.sleep(1)И пока программа крутится в этом цикле, она не будет делать ничего другого.
Если тебе нужно делать что-то параллельно с периодическим вызовом, то дело усложняется.
Если твоя программа синхронная, то тебе потребуется отдельный поток выполнения.
import threading class MyJob(threading.Thread): def __init__(self): super().__init__(self) self.stop = threading.Event() def proc_to_call(self): #Это твой периодический вызов pass def run(self): while not self.stop.wait(60): self.proc_to_call() job = MyJob() job.start() #код основного потока продолжает выполняться после этого #когда нужен останов, делаешь job.stop.set() #сигналим об останове job.join() #ждём завершения потокаПочему использую Event вместо простой логической переменной и time.sleep()? Потому что в случае с time.sleep() тебе придётся ждать конца минуты, прежде чем поток снова проверит логический флаг и поймёт, что пора останавливаться. А при использовании Event его установка тут же прервёт ожидание — и выйдет из цикла.
А если твоя программа асинхронная, то всё упрощается, так как корутины уже могут чередовать своё выполнение, выполняясь «как бы параллельно». Можно использовать как schedule, так и простой цикл, просто не забывай использовать await asyncio.sleep() вместо time.sleep().
Библиотека schedule – CRON на Python
Вам приходилось работать с CRON? Это такой сервис в nix-системах, который позволяет регулярно в определенные моменты времени запускать скрипты или программы. Штука с долгой историей, в наследство которой достался странный синтаксис для описания правил:
Что если бы мы хотели иметь свой CRON внутри программы Python, чтобы в нужные моменты времени вызывать функции? Да еще, чтобы у него был человеческий синтаксис? Такая библиотека есть и называется schedule.
import schedule import time def job(): print("Работаю") schedule.every(10).minutes.do(job) schedule.every().hour.do(job) schedule.every().day.at("10:30").do(job) schedule.every(5).to(10).minutes.do(job) schedule.every().monday.do(job) schedule.every().wednesday.at("13:15").do(job) schedule.every().minute.at(":17").do(job) # нужно иметь свой цикл для запуска планировщика с периодом в 1 секунду: while True: schedule.run_pending() time.sleep(1) Как видите, правила для задания временных интервалов прекрасно читаются, словно они предложения на английском языке. Перевод пары примеров:
# спланируй.каждые(10).минут.сделать(работу) schedule.every(10).minutes.do(job) # спланируй.каждый().день.в(10:30).сделать(работу) schedule.every().day.at("10:30").do(job) В задания можно передавать параметры вот так:
def greet(name): print('Hello', name) schedule.every(2).seconds.do(greet, name='Alice') Если по какой-то причине нужно отменить задание, это делается так:
def job1(): # возвращаем такой токен, и это задание снимается с выполнения в будущем return schedule.CancelJob schedule.every().day.at('22:30').do(job1) Если нужно отменить группу заданий, то к ним добавляют тэги:
schedule.every().day.do(greet, 'Monica').tag('daily-tasks') schedule.every().day.do(greet, 'Derek').tag('daily-tasks') schedule.clear('daily-tasks') # массовая отмена по тэгу Метод to позволяет задать случайный интервал для выполнения задания, например от 5 до 10 секунд:
schedule.every(5).to(10).seconds.do(my_job)
Библиотека сама не обрабатывает сама исключения в ваших задачах, поэтому, возможно, понадобится создать подкласс планировщика, как в этом примере. Или декоратор, который будет отменять работу, если произошло исключение. Вот так:
import functools # декоратор для ловли исключений def catch_exceptions(cancel_on_failure=False): def catch_exceptions_decorator(job_func): @functools.wraps(job_func) def wrapper(*args, **kwargs): try: return job_func(*args, **kwargs) except: import traceback print(traceback.format_exc()) if cancel_on_failure: return schedule.CancelJob return wrapper return catch_exceptions_decorator @catch_exceptions(cancel_on_failure=True) def bad_task(): # даст исключение, но декоратор просто отменит эту задачу return 1 / 0 schedule.every(5).minutes.do(bad_task)
Если задания занимают продолжительное время или должны выполняться параллельно, то вам самостоятельно придется организовать их выполнение в отдельных потоках.
import threading import time import schedule # код задания def job(): print("Выполняюсь в отдельном потоке") def run_threaded(job_func): job_thread = threading.Thread(target=job_func) job_thread.start() schedule.every(10).seconds.do(run_threaded, job) schedule.every(10).seconds.do(run_threaded, job) schedule.every(10).seconds.do(run_threaded, job) schedule.every(10).seconds.do(run_threaded, job) schedule.every(10).seconds.do(run_threaded, job) # бесконечный цикл, проверяющий каждую секунду, не пора ли запустить задание while 1: schedule.run_pending() time.sleep(1) Celery
Если вы используете в проекте Celery, то, вероятно, вам не нужен schedule. В Celery и так есть отличный CRON:
from celery import Celery from celery.schedules import crontab app = Celery() @app.on_after_configure.connect def setup_periodic_tasks(sender, **kwargs): # эта функция выполнится при запуске - настроим вызовы задачи test sender.add_periodic_task(10.0, test.s('hello'), name='add every 10') sender.add_periodic_task(30.0, test.s('world'), expires=10) # в 7 30 по понедельникам sender.add_periodic_task( crontab(hour=7, minute=30, day_of_week=1), test.s('Happy Mondays!'), ) @app.task def test(arg): print(arg) Специально для канала @pyway. Подписывайтесь на мой канал в Телеграм @pyway 👈