Блог

У меня есть веб-сервер, который создает файл при вызове. Я хотел бы добавить куда-нибудь функцию, запускаемую одновременно, которая проверяла бы этот файл и воздействовала на его содержимое, но я не знаю, где разместить его в коде. Код для веб-сервера:

 import bottle
import pickle
import time

class WebServer(object):
    def __init__(self):
        bottle.route("/", 'GET', self.root)
        bottle.run(host='0.0.0.0', debug=True)

    def root(self):
        with open("watchdog.txt", "wb") as f:
            pickle.dump(time.time(), f)

if __name__ == "__main__":
    WebServer()
  

Функция, которую я хотел бы запустить вместе с веб-сервером:

 def check():
    with open("watchdog.txt", "rb") as f:
        t1 = pickle.load(f)
    t2 = time.time()
    if t2 - t1 > 10:
        print("stale watchdog")
  

Вызов WebServer() помещает программу в цикл (что нормально, веб-сервер прослушивает), поэтому я хотел бы поместить check() куда-нибудь, где это можно было бы объединить с обратным вызовом (аналогично self.root.after() в Tkinter). Как лучше всего это сделать?

ПРИМЕЧАНИЕ: я пропустил в приведенном выше коде проверку ошибок, учет отсутствия watchdog.txt и т. Д. для простоты.

Одно из решений, которое я наконец нашел, — использовать планировщик событий:

 import bottle
import pickle
import time
import threading

class WebServer(object):
    def __init__(self):
        bottle.route("/", 'GET', self.root)
        bottle.run(host='0.0.0.0', debug=True)

    def root(self):
        with open("watchdog.txt", "wb") as f:
            pickle.dump(time.time(), f)


def check():
    try:
        with open("watchdog.txt", "rb") as f:
            t1 = pickle.load(f)
    except IOError:
        pass
    else:
        t2 = time.time()
        if t2 - t1 > 10:
            print("stale watchdog")
        else:
            print("fresh watchdog")
    finally:
        threading.Timer(10, check).start()


if __name__ == "__main__":
    check()
    WebServer()