searxng/searx/shared/shared_uwsgi.py
Markus Heiser 3d96a9839a [format.python] initial formatting of the python code
This patch was generated by black [1]::

    make format.python

[1] https://github.com/psf/black

Signed-off-by: Markus Heiser <markus.heiser@darmarit.de>
2021-12-27 09:26:22 +01:00

63 lines
1.6 KiB
Python

# SPDX-License-Identifier: AGPL-3.0-or-later
import time
import uwsgi # pylint: disable=E0401
from . import shared_abstract
_last_signal = 10
class UwsgiCacheSharedDict(shared_abstract.SharedDict):
def get_int(self, key):
value = uwsgi.cache_get(key)
if value is None:
return value
else:
return int.from_bytes(value, 'big')
def set_int(self, key, value):
b = value.to_bytes(4, 'big')
uwsgi.cache_update(key, b)
def get_str(self, key):
value = uwsgi.cache_get(key)
if value is None:
return value
else:
return value.decode('utf-8')
def set_str(self, key, value):
b = value.encode('utf-8')
uwsgi.cache_update(key, b)
def schedule(delay, func, *args):
"""
Can be implemented using a spooler.
https://uwsgi-docs.readthedocs.io/en/latest/PythonDecorators.html
To make the uwsgi configuration simple, use the alternative implementation.
"""
global _last_signal
def sighandler(signum):
now = int(time.time())
key = 'scheduler_call_time_signal_' + str(signum)
uwsgi.lock()
try:
updating = uwsgi.cache_get(key)
if updating is not None:
updating = int.from_bytes(updating, 'big')
if now - updating < delay:
return
uwsgi.cache_update(key, now.to_bytes(4, 'big'))
finally:
uwsgi.unlock()
func(*args)
signal_num = _last_signal
_last_signal += 1
uwsgi.register_signal(signal_num, 'worker', sighandler)
uwsgi.add_timer(signal_num, delay)
return True