python как создать декоратор для планировщика Schedule

Question

Как возможно создать декоратор для планировщика задач : Schedule Наподобие такого: запуск задачи каждые 20 секунд

@schedule(seconds=2)

def my_function(self):
    # -- основной код программы -- #

""" Пример реализации Планировщика
когда функция выполняется по расписанию (заускается каждый день например) 
(в отдельном потоке) """

import schedule
import time
import threading
from threading import Thread
import datetime


def test():
    a = 12
    b = a + 8
    print(b)

def geeks():
    print(f"Ready and datetime_now :{datetime.datetime.now()}\n")
    thread_name = threading.current_thread().name
    print(f"Thread_name: {thread_name}\n")

def geeks_2():
    print(f"\nWoW + datetime_now : {datetime.datetime.now()}\n")
    thread_name = threading.current_thread().name
    print(f"Thread_name: {thread_name}\n")

def run():
    schedule.every(10).seconds.do(geeks)

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

def run_2():
    schedule.every(20).seconds.do(geeks_2)

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

# thread = Thread(name='worker', target=run)
# thread.start()



def run_thread():
    thread_2 = Thread(name='worker_2', target=run_2)
    thread_2.start()

run_thread()

test()

Answer 1

В отличие от обычных декораторов тут не нужно оборачивать функцию во wrapper, просто возвращаем ее же, но перед этим создаем объект Job() с помощью функции every(), и вызываем job.do(func).

import datetime
import schedule
import time
import threading


def run_pending():
    while True:
        schedule.run_pending()
        time.sleep(0.1)


# daemon=True значит что этот поток будет работать, пока работает пока работает основной поток
threading.Thread(target=run_pending, daemon=True).start()


def scheduler(**interval_args):
    # Получаем имя и значение параметра, например unit='seconds'; interval=10
    unit, interval = list(interval_args.items())[0]  # Параметр должен быть обязательно передан, иначе будет IndexError
    job = schedule.every(interval)

    # Немного некорректно прописывать `unit` напрямую в обход свойств `.seconds`/`.minutes`/`.hour` и т.д.,
    # но работать тем не менее будет
    job.unit = unit  

    def decorator(func):
        job.do(func)
        return func  # Возвращаем саму функцию, никак не заменяя
    
    return decorator


# ----


@scheduler(seconds=1)  # единица_измерения=интервал
def geeks():
    print("geeks")
    print(f"Ready and datetime_now :{datetime.datetime.now()}")
    thread_name = threading.current_thread().name
    print(f"Thread_name: {thread_name}")
    print()


@scheduler(seconds=2)
def geeks_2():
    print("geeks2")
    print(f"\nWoW + datetime_now : {datetime.datetime.now()}")
    thread_name = threading.current_thread().name
    print(f"Thread_name: {thread_name}")
    print()


while True:
    time.sleep(10)

Можно упростить код декоратора, если передавать в него параметром объект Job, который возвращается функцией every() (только не добавлять в конце .do()), тогда синтаксис задания интервала будет "унаследован" от библиотеки schedule:

def schedule_job(job: schedule.Job):
    def decorator(func):
        job.do(func)
        return func
    
    return decorator


# ----
# Для укорочения записи интервалов импортируем every
from schedule import every


@schedule_job(every().second)
def geeks():
    print("geeks")
    print(f"Ready and datetime_now :{datetime.datetime.now()}")
    thread_name = threading.current_thread().name
    print(f"Thread_name: {thread_name}")
    print()


@schedule_job(every(2).seconds)
def geeks_2():
    print("geeks2")
    print(f"\nWoW + datetime_now : {datetime.datetime.now()}")
    thread_name = threading.current_thread().name
    print(f"Thread_name: {thread_name}")
    print()

И аналогично @schedule_job(every().week), @schedule_job(every().wednesday.at("13:15")), и т.д.

UPD. оказалось, в библиотеке schedule уже есть свой декоратор, причем на 90% процентов совпадает с моим вторым вариантом (в отличие от моего варианта есть еще возможность передачи параметров из декоратора в декорируемую функцию):

Источник, соответствующий pull request, и пример в документации

def repeat(job, *args, **kwargs):
    """
    Decorator to schedule a new periodic job.
    Any additional arguments are passed on to the decorated function
    when the job runs.
    :param job: a :class:`Jobs <Job>`
    """

    def _schedule_decorator(decorated_function):
        job.do(decorated_function, *args, **kwargs)
        return decorated_function

    return _schedule_decorator

БЛОГ НА HUSL

python как создать декоратор для планировщика Schedule

Ответы (1 шт):