Модуль [ schedule ] [ Python 3 ]

4.0/5 оценка (2 голосов)

Материал:
Завершен
Обновлен: 15 января 2023
[ Python 3 ]

Пакет schedule

schedule позволяет планировать выполнение каких-либо задач на определённый промежуток времени.

Пакет мало весит, имеет ясный и интуитивно понятный функционал.

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

Документация доступна по ссылке:
https://schedule.readthedocs.io/en/stable/

Examples

Эта страница дает представление о модуле schedule. Предполагается, что он у вас уже установлен. Если нет, то вот ссылка: Installation.

Выполнение задания job каждые x минут:

import schedule
import time

def job():
    print("I'm working...")

# Выполняйте задание каждую 3 секунду / минуту/ час / день / неделю,
# Начиная с 3 секунды / минуты / часа / дня /недели от этого момента
schedule.every(3).seconds.do(job)
schedule.every(3).minutes.do(job)
schedule.every(3).hours.do(job)
schedule.every(3).days.do(job)
schedule.every(3).weeks.do(job)

# Выполняйте задание каждую минуту на 23-й секунде
schedule.every().minute.at(":23").do(job)

# Выполняйте задание каждый час на 42-й минуте
schedule.every().hour.at(":42").do(job)

# Выполняйте задания каждый 5 час, в 20 минут и 30 секунд.
# Если текущее время 02:00, то первое выполнение состоится в 06:20:30
schedule.every(5).hours.at("20:30").do(job)

# Выполняйте задание каждый день в определенное время ЧЧ:ММ и следующее ЧЧ:ММ:СС
schedule.every().day.at("10:30").do(job)
schedule.every().day.at("10:30:42").do(job)

# Выполнить задание в определенный день недели
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)
schedule.every().minute.at(":17").do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

Используйте декоратор, чтобы запланировать работу

Используйте @repeat, чтобы запланировать функцию. Передайте ему интервал, используя тот же синтаксис, что и выше, но опуская .do().

from schedule import every, repeat, run_pending
import time

@repeat(every(10).minutes)
def job():
    print("I am a scheduled job")

while True:
    run_pending()
    time.sleep(1)

@repeat декоратор не работает с нестатическими методами класса.

Передача аргументов в функцию job

do() передает дополнительные аргументы функции job

import schedule

def greet(name):
    print('Hello', name)

schedule.every(2).seconds.do(greet, name='Alice')
schedule.every(4).seconds.do(greet, name='Bob')

from schedule import every, repeat

@repeat(every().second, "World")
@repeat(every().day, "Mars")
def hello(planet):
    print("Hello", planet)

Завершение работы функции job

Чтобы удалить задание из планировщика, используйте метод schedule.cancel_job(job)

import schedule

def some_task():
    print('Hello world')

job = schedule.every().day.at('22:30').do(some_task)
schedule.cancel_job(job)

Выполнить задание один раз

Возвратите schedule.CancelJob из задания, чтобы удалить его из планировщика.

import schedule
import time

def job_that_executes_once():
    # Сделайте какую-нибудь работу, которая должна произойти только один раз...
    return schedule.CancelJob

schedule.every().day.at('22:30').do(job_that_executes_once)

while True:
    schedule.run_pending()
    time.sleep(1)

Извлечь все задания

Чтобы извлечь все задания из планировщика, используйте schedule.get_jobs()

import schedule

def hello():
    print('Hello world')

schedule.every().second.do(hello)

all_jobs = schedule.get_jobs()

Отменить все задания

Чтобы удалить все задания из планировщика, используйте schedule.clear()

import schedule

def greet(name):
    print('Hello {}'.format(name))

schedule.every().second.do(greet)

schedule.clear()

Получите несколько заданий, отфильтрованных по тегам

Вы можете извлечь группу заданий из планировщика, выбрав их по уникальному идентификатору.

import schedule

def greet(name):
    print('Hello {}'.format(name))

schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend')
schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend')
schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')
schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest')

friends = schedule.get_jobs('friend')

Вернет список всех заданий, помеченных как friend.

Отменить несколько заданий, отфильтрованных по тегам

Вы можете отменить планирование группы заданий, выбрав их по уникальному идентификатору.

import schedule

def greet(name):
    print('Hello {}'.format(name))

schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend')
schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend')
schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')
schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest')

schedule.clear('daily-tasks')

Предотвратит каждое задание, помеченное как daily-tasks для повторного запуска.

Выполняйте задание через случайные промежутки времени

def my_job():
    print('Foo')

# Run every 5 to 10 seconds.
schedule.every(5).to(10).seconds.do(my_job)

every(A).to(B).seconds executes the job function every N seconds such that A <= N <= B.

Выполняйте задание до определенного времени

import schedule
from datetime import datetime, timedelta, time

def job():
    print('Boo')

# run job until a 18:30 today
schedule.every(1).hours.until("18:30").do(job)

# run job until a 2030-01-01 18:33 today
schedule.every(1).hours.until("2030-01-01 18:33").do(job)

# Schedule a job to run for the next 8 hours
schedule.every(1).hours.until(timedelta(hours=8)).do(job)

# Run my_job until today 11:33:42
schedule.every(1).hours.until(time(11, 33, 42)).do(job)

# run job until a specific datetime
schedule.every(1).hours.until(datetime(2020, 5, 17, 11, 36, 20)).do(job)

until метод устанавливает крайний срок выполнения заданий. Задание не будет выполнено после истечения крайнего срока.

Время до следующего выполнения

Используйте schedule.idle_seconds() чтобы узнать количество секунд до запланированного выполнения следующего задания. Возвращаемое значение будет отрицательным, если выполнение следующих запланированных заданий было запланировано в прошлом. Returns None если никакие задания не запланированы.

import schedule
import time

def job():
    print('Hello')

schedule.every(5).seconds.do(job)

while 1:
    n = schedule.idle_seconds()
    if n is None:
        # no more jobs
        break
    elif n > 0:
        # sleep exactly the right amount of time
        time.sleep(n)
    schedule.run_pending()

Выполняйте все задания сейчас, независимо от их расписания

Чтобы запустить все задания независимо от того, запланированы они к запуску или нет, используйте schedule.run_all(). Задания будут после завершения повторно планироваться, точно так же, как если бы они выполнялись с использованием run_pending().

import schedule

def job_1():
    print('Foo')

def job_2():
    print('Bar')

schedule.every().monday.at("12:40").do(job_1)
schedule.every().tuesday.at("16:40").do(job_2)

schedule.run_all()

# Добавьте аргумент delay_seconds для запуска заданий с
# 10 секундной задержкой между ними.
schedule.run_all(delay_seconds=10)

Запуск в фоновом режиме

Из коробки невозможно запустить schedule в фоновом режиме. Однако возможно самостоятельно создать поток и использовать его для выполнения заданий, не блокируя основной поток.

Например:

import threading
import time

import schedule


def run_continuously(interval=1):
    """Continuously run, while executing pending jobs at each
    elapsed time interval.
    @return cease_continuous_run: threading. Event which can
    be set to cease continuous run. Please note that it is
    *intended behavior that run_continuously() does not run
    missed jobs*. For example, if you've registered a job that
    should run every minute and you set a continuous run
    interval of one hour then your job won't be run 60 times
    at each interval but only once.
    """
    cease_continuous_run = threading.Event()

    class ScheduleThread(threading.Thread):
        @classmethod
        def run(cls):
            while not cease_continuous_run.is_set():
                schedule.run_pending()
                time.sleep(interval)

    continuous_thread = ScheduleThread()
    continuous_thread.start()
    return cease_continuous_run


def background_job():
    print('Hello from the background thread')


schedule.every().second.do(background_job)

# Start the background thread
stop_run_continuously = run_continuously()

# Do some other things...
time.sleep(10)

# Stop the background thread
stop_run_continuously.set()