50 lines
1.3 KiB
Python
50 lines
1.3 KiB
Python
import threading
|
|
from watchdog.events import PatternMatchingEventHandler
|
|
from watchdog.observers import Observer
|
|
import os
|
|
import time
|
|
|
|
FLAGS_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
|
|
|
|
class FlagHandler(PatternMatchingEventHandler):
|
|
def __init__(self, shared_state, lock):
|
|
super().__init__(patterns=["*.flag"], ignore_directories=True)
|
|
self.shared_state = shared_state
|
|
self.lock = lock
|
|
|
|
def on_created(self, event):
|
|
try:
|
|
with open(event.src_path, "r") as f:
|
|
content = f.read().strip()
|
|
except FileNotFoundError:
|
|
return
|
|
try:
|
|
os.remove(event.src_path)
|
|
except FileNotFoundError:
|
|
pass
|
|
with self.lock:
|
|
self.shared_state["current"] = content
|
|
time.sleep(2)
|
|
|
|
shared_state = {"current": None}
|
|
lock = threading.Lock()
|
|
|
|
observer = Observer()
|
|
observer.schedule(FlagHandler(shared_state, lock), path=FLAGS_DIR, recursive=False)
|
|
observer.start()
|
|
|
|
def get_status():
|
|
with lock:
|
|
current = shared_state["current"]
|
|
if current == "db_check_done":
|
|
return False
|
|
return current
|
|
|
|
def get_migration():
|
|
with lock:
|
|
current = shared_state["current"]
|
|
if current == "migration_done":
|
|
return True
|
|
return False |