forked from Archives/searxng
6e2872f436
shared dictionary between the workers (UWSGI or werkzeug) scheduler: run a task once every x seconds (UWSGI or werkzeug)
39 lines
698 B
Python
39 lines
698 B
Python
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
|
|
import threading
|
|
|
|
from . import shared_abstract
|
|
|
|
|
|
class SimpleSharedDict(shared_abstract.SharedDict):
|
|
|
|
__slots__ = 'd',
|
|
|
|
def __init__(self):
|
|
self.d = {}
|
|
|
|
def get_int(self, key):
|
|
return self.d.get(key, None)
|
|
|
|
def set_int(self, key, value):
|
|
self.d[key] = value
|
|
|
|
def get_str(self, key):
|
|
return self.d.get(key, None)
|
|
|
|
def set_str(self, key, value):
|
|
self.d[key] = value
|
|
|
|
|
|
def schedule(delay, func, *args):
|
|
def call_later():
|
|
t = threading.Timer(delay, wrapper)
|
|
t.daemon = True
|
|
t.start()
|
|
|
|
def wrapper():
|
|
call_later()
|
|
func(*args)
|
|
|
|
call_later()
|