Расписание повторяющееся событие в Python 3

Question

Я пытаюсь запланировать повторяющееся событие, чтобы запустить каждую минуту в Python 3.

Я видел класс sched.scheduler, но мне интересно, если есть другой способ сделать это. Я слышал упоминания, что я мог бы использовать несколько потоков для этого, что я бы не прочь сделать.

Я в основном запрашиваю JSON и затем разбираю его; его стоимость меняется со временем.

Для использования sched.scheduler я должен создать цикл, чтобы просить его запланировать даже работать в течение одного часа: там

scheduler = sched.scheduler(time.time, time.sleep) 

# Schedule the event. THIS IS UGLY! 
for i in range(60): 
    scheduler.enter(3600 * i, 1, query_rate_limit,()) 

scheduler.run() 

Какие другие способы сделать это будут?

Answer 1

Вы можете использовать threading.Timer, но это также предусматривает одноразовое событие, аналогично методу планировщика .enter.

Нормальный шаблон (на любом языке) для преобразования разового планировщика в периодический планировщик состоит в том, чтобы каждое событие перепланировало себя в указанный интервал. Например, с sched, я бы не использовать цикл, как вы делаете, а скорее что-то вроде:

def periodic(scheduler, interval, action, actionargs=()): 
    scheduler.enter(interval, 1, periodic, 
        (scheduler, interval, action, actionargs)) 
    action(*actionargs) 

и начать весь «навсегда периодический график» с вызовом

periodic(scheduler, 3600, query_rate_limit) 

Или , Я мог бы использовать threading.Timer вместо scheduler.enter, но шаблон очень похож.

Если вам требуется более тонкая вариация (например, остановить периодическую перепланировку в заданное время или при определенных условиях), это не слишком сложно для размещения с несколькими дополнительными параметрами.

Answer 2

Мое скромное взятие на тему:

from threading import Timer 

class RepeatedTimer(object): 
    def __init__(self, interval, function, *args, **kwargs): 
     self._timer  = None 
     self.function = function 
     self.interval = interval 
     self.args  = args 
     self.kwargs  = kwargs 
     self.is_running = False 
     self.start() 

    def _run(self): 
     self.is_running = False 
     self.start() 
     self.function(*self.args, **self.kwargs) 

    def start(self): 
     if not self.is_running: 
      self._timer = Timer(self.interval, self._run) 
      self._timer.start() 
      self.is_running = True 

    def stop(self): 
     self._timer.cancel() 
     self.is_running = False 

Использование:

from time import sleep 

def hello(name): 
    print "Hello %s!" % name 

print "starting..." 
rt = RepeatedTimer(1, hello, "World") # it auto-starts, no need of rt.start() 
try: 
    sleep(5) # your long-running job goes here... 
finally: 
    rt.stop() # better in a try/finally block to make sure the program ends! 

Особенности:

только
• Стандартная библиотека, никаких внешних зависимостей
• Использует шаблон, предложенный Алекс Мартнелли
• start() и stop() безопасны для вызова несколько раз, даже если таймер уже запущен/остановлена ​​
• функции вызываемой может иметь позиционные и именованные аргументы
• Вы можете изменить interval в любое время, она будет действовать после следующего запуска. То же самое для args, kwargs и даже function!

Answer 3

Вы можете использовать Advanced Python Scheduler. У этого даже есть cron-подобный интерфейс.

Answer 4

Вы можете использовать schedule. Он работает на Python 2.7 и 3.3 и довольно легкий:

import schedule 
import time 

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

schedule.every(10).minutes.do(job) 
schedule.every().hour.do(job) 
schedule.every().day.at("10:30").do(job) 

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

Answer 5

основе MestreLion ответа, это решить небольшую проблему с многопоточностью:

from threading import Timer, Lock 


class Periodic(object): 
    """ 
    A periodic task running in threading.Timers 
    """ 

    def __init__(self, interval, function, *args, **kwargs): 
     self._lock = Lock() 
     self._timer = None 
     self.function = function 
     self.interval = interval 
     self.args = args 
     self.kwargs = kwargs 
     self._stopped = True 
     if kwargs.pop('autostart', True): 
      self.start() 

    def start(self, from_run=False): 
     self._lock.acquire() 
     if from_run or self._stopped: 
      self._stopped = False 
      self._timer = Timer(self.interval, self._run) 
      self._timer.start() 
      self._lock.release() 

    def _run(self): 
     self.start(from_run=True) 
     self.function(*self.args, **self.kwargs) 

    def stop(self): 
     self._lock.acquire() 
     self._stopped = True 
     self._timer.cancel() 
     self._lock.release() 

Answer 6

Использование Celery.

from celery.task import PeriodicTask 
from datetime import timedelta 


class ProcessClicksTask(PeriodicTask): 
    run_every = timedelta(minutes=30) 

    def run(self, **kwargs): 
     #do something 

Answer 7

Задача-планировщик является планировщик в процессе, чтобы организовать и запустить задачу периодически в соответствии с конфигурационный файл YAML. Пожалуйста, обратитесь к https://github.com/tzutalin/task-scheduler

Answer 8

На основании ответа Алекса Мартелли, я внедрил декоратор версии, которая более легче интегрироваться.

import sched 
import time 
import datetime 
from functools import wraps 
from threading import Thread 


def async(func): 
    @wraps(func) 
    def async_func(*args, **kwargs): 
     func_hl = Thread(target=func, args=args, kwargs=kwargs) 
     func_hl.start() 
     return func_hl 
    return async_func 


def schedule(interval): 
    def decorator(func): 
     def periodic(scheduler, interval, action, actionargs=()): 
      scheduler.enter(interval, 1, periodic, 
          (scheduler, interval, action, actionargs)) 
      action(*actionargs) 

     @wraps(func) 
     def wrap(*args, **kwargs): 
      scheduler = sched.scheduler(time.time, time.sleep) 
      periodic(scheduler, interval, func) 
      scheduler.run() 
     return wrap 
    return decorator 


@async 
@schedule(1) 
def periodic_event(): 
    print(datetime.datetime.now()) 


if __name__ == '__main__': 
    print('start') 
    periodic_event() 
    print('end')