diff options
| -rw-r--r-- | searx/exceptions.py | 5 | ||||
| -rw-r--r-- | searx/search/checker/background.py | 72 | ||||
| -rw-r--r-- | searx/shared/shared_simple.py | 9 | ||||
| -rw-r--r-- | searx/shared/shared_uwsgi.py | 9 |
4 files changed, 73 insertions, 22 deletions
diff --git a/searx/exceptions.py b/searx/exceptions.py index 1b106d40c..43c8bab40 100644 --- a/searx/exceptions.py +++ b/searx/exceptions.py @@ -16,6 +16,9 @@ along with searx. If not, see < http://www.gnu.org/licenses/ >. ''' +from typing import Optional, Union + + class SearxException(Exception): pass @@ -35,7 +38,7 @@ class SearxParameterException(SearxException): class SearxSettingsException(SearxException): """Error while loading the settings""" - def __init__(self, message, filename): + def __init__(self, message: Union[str, Exception], filename: Optional[str]): super().__init__(message) self.message = message self.filename = filename diff --git a/searx/search/checker/background.py b/searx/search/checker/background.py index ff005dd91..f47e7d752 100644 --- a/searx/search/checker/background.py +++ b/searx/search/checker/background.py @@ -1,6 +1,7 @@ # SPDX-License-Identifier: AGPL-3.0-or-later # lint: pylint # pylint: disable=missing-module-docstring +# pyright: strict import json import random @@ -8,6 +9,8 @@ import time import threading import os import signal +from typing import Dict, Union, List, Any, Tuple +from typing_extensions import TypedDict, Literal from searx import logger, settings, searx_debug from searx.exceptions import SearxSettingsException @@ -20,17 +23,58 @@ CHECKER_RESULT = 'CHECKER_RESULT' running = threading.Lock() -def _get_interval(every, error_msg): +CheckerResult = Union['CheckerOk', 'CheckerErr', 'CheckerOther'] + + +class CheckerOk(TypedDict): + """Checking the engines succeeded""" + + status: Literal['ok'] + engines: Dict[str, 'EngineResult'] + timestamp: int + + +class CheckerErr(TypedDict): + """Checking the engines failed""" + + status: Literal['error'] + timestamp: int + + +class CheckerOther(TypedDict): + """The status is unknown or disabled""" + + status: Literal['unknown', 'disabled'] + + +EngineResult = Union['EngineOk', 'EngineErr'] + + +class EngineOk(TypedDict): + """Checking the engine succeeded""" + + success: Literal[True] + + +class EngineErr(TypedDict): + """Checking the engine failed""" + + success: Literal[False] + errors: Dict[str, List[str]] + + +def _get_interval(every: Any, error_msg: str) -> Tuple[int, int]: if isinstance(every, int): - every = (every, every) + return (every, every) + if ( not isinstance(every, (tuple, list)) - or len(every) != 2 + or len(every) != 2 # type: ignore or not isinstance(every[0], int) or not isinstance(every[1], int) ): raise SearxSettingsException(error_msg, None) - return every + return (every[0], every[1]) def _get_every(): @@ -38,25 +82,27 @@ def _get_every(): return _get_interval(every, 'checker.scheduling.every is not a int or list') -def get_result(): +def get_result() -> CheckerResult: serialized_result = storage.get_str(CHECKER_RESULT) if serialized_result is not None: return json.loads(serialized_result) return {'status': 'unknown'} -def _set_result(result, include_timestamp=True): - if include_timestamp: - result['timestamp'] = int(time.time() / 3600) * 3600 +def _set_result(result: CheckerResult): storage.set_str(CHECKER_RESULT, json.dumps(result)) +def _timestamp(): + return int(time.time() / 3600) * 3600 + + def run(): if not running.acquire(blocking=False): # pylint: disable=consider-using-with return try: logger.info('Starting checker') - result = {'status': 'ok', 'engines': {}} + result: CheckerOk = {'status': 'ok', 'engines': {}, 'timestamp': _timestamp()} for name, processor in PROCESSORS.items(): logger.debug('Checking %s engine', name) checker = Checker(processor) @@ -69,7 +115,7 @@ def run(): _set_result(result) logger.info('Check done') except Exception: # pylint: disable=broad-except - _set_result({'status': 'error'}) + _set_result({'status': 'error', 'timestamp': _timestamp()}) logger.exception('Error while running the checker') finally: running.release() @@ -89,7 +135,7 @@ def _start_scheduling(): run() -def _signal_handler(_signum, _frame): +def _signal_handler(_signum: int, _frame: Any): t = threading.Thread(target=run) t.daemon = True t.start() @@ -102,7 +148,7 @@ def initialize(): signal.signal(signal.SIGUSR1, _signal_handler) # disabled by default - _set_result({'status': 'disabled'}, include_timestamp=False) + _set_result({'status': 'disabled'}) # special case when debug is activate if searx_debug and settings.get('checker', {}).get('off_when_debug', True): @@ -116,7 +162,7 @@ def initialize(): return # - _set_result({'status': 'unknown'}, include_timestamp=False) + _set_result({'status': 'unknown'}) start_after = scheduling.get('start_after', (300, 1800)) start_after = _get_interval(start_after, 'checker.scheduling.start_after is not a int or list') diff --git a/searx/shared/shared_simple.py b/searx/shared/shared_simple.py index 0bf13a2a6..2b9d4c2da 100644 --- a/searx/shared/shared_simple.py +++ b/searx/shared/shared_simple.py @@ -1,6 +1,7 @@ # SPDX-License-Identifier: AGPL-3.0-or-later import threading +from typing import Optional from . import shared_abstract @@ -12,16 +13,16 @@ class SimpleSharedDict(shared_abstract.SharedDict): def __init__(self): self.d = {} - def get_int(self, key): + def get_int(self, key: str) -> Optional[int]: return self.d.get(key, None) - def set_int(self, key, value): + def set_int(self, key: str, value: int): self.d[key] = value - def get_str(self, key): + def get_str(self, key: str) -> Optional[str]: return self.d.get(key, None) - def set_str(self, key, value): + def set_str(self, key: str, value: str): self.d[key] = value diff --git a/searx/shared/shared_uwsgi.py b/searx/shared/shared_uwsgi.py index 592e24a4b..4a6b0a155 100644 --- a/searx/shared/shared_uwsgi.py +++ b/searx/shared/shared_uwsgi.py @@ -1,6 +1,7 @@ # SPDX-License-Identifier: AGPL-3.0-or-later import time +from typing import Optional import uwsgi # pylint: disable=E0401 from . import shared_abstract @@ -9,25 +10,25 @@ _last_signal = 10 class UwsgiCacheSharedDict(shared_abstract.SharedDict): - def get_int(self, key): + def get_int(self, key: str) -> Optional[int]: value = uwsgi.cache_get(key) if value is None: return value else: return int.from_bytes(value, 'big') - def set_int(self, key, value): + def set_int(self, key: str, value: int): b = value.to_bytes(4, 'big') uwsgi.cache_update(key, b) - def get_str(self, key): + def get_str(self, key: str) -> Optional[str]: value = uwsgi.cache_get(key) if value is None: return value else: return value.decode('utf-8') - def set_str(self, key, value): + def set_str(self, key: str, value: str): b = value.encode('utf-8') uwsgi.cache_update(key, b) |