Merge pull request #817 from not-my-profile/pyright-01

Pyright 01

package.json

@@ -1,6 +1,6 @@
 {
   "dependencies": {
     "eslint": "^8.7.0",
-    "pyright": "^1.1.212"
+    "pyright": "^1.1.215"
   }
 }

requirements.txt

@@ -15,3 +15,4 @@ langdetect==1.0.9
 setproctitle==1.2.2
 redis==4.1.1
 mistletoe==0.8.1
+typing_extensions==4.0.1

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

searx/network/__init__.py

@@ -7,6 +7,7 @@ import threading
 import concurrent.futures
 from types import MethodType
 from timeit import default_timer
+from typing import Iterable, Tuple
 
 import httpx
 import anyio
@@ -210,7 +211,7 @@ def _close_response_method(self):
         continue
 
 
-def stream(method, url, **kwargs):
+def stream(method, url, **kwargs) -> Tuple[httpx.Response, Iterable[bytes]]:
     """Replace httpx.stream.
 
     Usage:

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')

searx/shared/shared_abstract.py

@@ -1,20 +1,22 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
+# pyright: strict
 from abc import ABC, abstractmethod
+from typing import Optional
 
 
 class SharedDict(ABC):
     @abstractmethod
-    def get_int(self, key):
+    def get_int(self, key: str) -> Optional[int]:
         pass
 
     @abstractmethod
-    def set_int(self, key, value):
+    def set_int(self, key: str, value: int):
         pass
 
     @abstractmethod
-    def get_str(self, key):
+    def get_str(self, key: str) -> Optional[str]:
         pass
 
     @abstractmethod
-    def set_str(self, key, value):
+    def set_str(self, key: str, value: str):
         pass

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
 
 

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)
 

searx/webapp.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 # SPDX-License-Identifier: AGPL-3.0-or-later
 # lint: pylint
+# pyright: basic
 """WebbApp
 
 """
@@ -33,11 +34,11 @@ from flask import (
     Flask,
     render_template,
     url_for,
-    Response,
     make_response,
     redirect,
     send_from_directory,
 )
+from flask.wrappers import Response
 from flask.ctx import has_request_context
 from flask.json import jsonify
 
@@ -255,7 +256,13 @@ flask_babel.get_translations = _get_translations
 
 @babel.localeselector
 def get_locale():
-    locale = request.preferences.get_value('locale') if has_request_context() else 'en'
+    locale = 'en'
+
+    if has_request_context():
+        value = request.preferences.get_value('locale')
+        if value:
+            locale = value
+
     if locale == 'oc':
         request.form['use-translation'] = 'oc'
         locale = 'fr_FR'
@@ -310,6 +317,7 @@ def code_highlighter(codelines, language=None):
     html_code = ''
     tmp_code = ''
     last_line = None
+    line_code_start = None
 
     # parse lines
     for line, code in codelines:
@@ -351,9 +359,11 @@ def get_current_theme_name(override: str = None) -> str:
     if override and (override in themes or override == '__common__'):
         return override
     theme_name = request.args.get('theme', request.preferences.get_value('theme'))
-    if theme_name not in themes:
-        theme_name = default_theme
-    return theme_name
+
+    if theme_name and theme_name in themes:
+        return theme_name
+
+    return default_theme
 
 
 def get_result_template(theme_name: str, template_name: str):
@@ -380,7 +390,7 @@ def proxify(url: str):
     if not settings.get('result_proxy'):
         return url
 
-    url_params = dict(mortyurl=url.encode())
+    url_params = dict(mortyurl=url)
 
     if settings['result_proxy'].get('key'):
         url_params['mortyhash'] = hmac.new(settings['result_proxy']['key'], url.encode(), hashlib.sha256).hexdigest()
@@ -1140,7 +1150,8 @@ def image_proxy():
     def close_stream():
         nonlocal resp, stream
         try:
-            resp.close()
+            if resp:
+                resp.close()
             del resp
             del stream
         except httpx.HTTPError as e:
@@ -1207,7 +1218,7 @@ def stats():
     reverse, key_name, default_value = STATS_SORT_PARAMETERS[sort_order]
 
     def get_key(engine_stat):
-        reliability = engine_reliabilities.get(engine_stat['name']).get('reliablity', 0)
+        reliability = engine_reliabilities.get(engine_stat['name'], {}).get('reliablity', 0)
         reliability_order = 0 if reliability else 1
         if key_name == 'reliability':
             key = reliability