summaryrefslogtreecommitdiff
path: root/searx/search
diff options
context:
space:
mode:
authorMarkus Heiser <markus.heiser@darmarit.de>2025-08-22 17:17:51 +0200
committerMarkus Heiser <markus.heiser@darmarIT.de>2025-09-03 13:37:36 +0200
commit57b9673efb1b4fd18a3ac15e26da642201e2cd33 (patch)
tree79d3ecd365a1669a1109aa7e5dd3636bc1041d96 /searx/search
parent09500459feffa414dc7a0601bdb164464a8b0454 (diff)
[mod] addition of various type hints / tbc
- pyright configuration [1]_ - stub files: types-lxml [2]_ - addition of various type hints - enable use of new type system features on older Python versions [3]_ - ``.tool-versions`` - set python to lowest version we support (3.10.18) [4]_: Older versions typically lack some typing features found in newer Python versions. Therefore, for local type checking (before commit), it is necessary to use the older Python interpreter. .. [1] https://docs.basedpyright.com/v1.20.0/configuration/config-files/ .. [2] https://pypi.org/project/types-lxml/ .. [3] https://typing-extensions.readthedocs.io/en/latest/# .. [4] https://mise.jdx.dev/configuration.html#tool-versions Signed-off-by: Markus Heiser <markus.heiser@darmarit.de> Format: reST
Diffstat (limited to 'searx/search')
-rw-r--r--searx/search/__init__.py56
-rw-r--r--searx/search/processors/__init__.py36
-rw-r--r--searx/search/processors/abstract.py35
3 files changed, 70 insertions, 57 deletions
diff --git a/searx/search/__init__.py b/searx/search/__init__.py
index 36903b194..3ea33ff12 100644
--- a/searx/search/__init__.py
+++ b/searx/search/__init__.py
@@ -2,7 +2,9 @@
# pylint: disable=missing-module-docstring, too-few-public-methods
# the public namespace has not yet been finally defined ..
-# __all__ = ["EngineRef", "SearchQuery"]
+# __all__ = [..., ]
+
+import typing as t
import threading
from timeit import default_timer
@@ -15,21 +17,27 @@ from searx import settings
import searx.answerers
import searx.plugins
from searx.engines import load_engines
-from searx.extended_types import SXNG_Request
from searx.external_bang import get_bang_url
-from searx.metrics import initialize as initialize_metrics, counter_inc, histogram_observe_time
+from searx.metrics import initialize as initialize_metrics, counter_inc
from searx.network import initialize as initialize_network, check_network_configuration
from searx.results import ResultContainer
from searx.search.checker import initialize as initialize_checker
-from searx.search.models import SearchQuery
from searx.search.processors import PROCESSORS, initialize as initialize_processors
-from .models import EngineRef, SearchQuery
+
+if t.TYPE_CHECKING:
+ from .models import SearchQuery
+ from searx.extended_types import SXNG_Request
logger = logger.getChild('search')
-def initialize(settings_engines=None, enable_checker=False, check_network=False, enable_metrics=True):
+def initialize(
+ settings_engines: list[dict[str, t.Any]] = None, # pyright: ignore[reportArgumentType]
+ enable_checker: bool = False,
+ check_network: bool = False,
+ enable_metrics: bool = True,
+):
settings_engines = settings_engines or settings['engines']
load_engines(settings_engines)
initialize_network(settings_engines, settings['outgoing'])
@@ -44,27 +52,25 @@ def initialize(settings_engines=None, enable_checker=False, check_network=False,
class Search:
"""Search information container"""
- __slots__ = "search_query", "result_container", "start_time", "actual_timeout"
+ __slots__ = "search_query", "result_container", "start_time", "actual_timeout" # type: ignore
- def __init__(self, search_query: SearchQuery):
+ def __init__(self, search_query: "SearchQuery"):
"""Initialize the Search"""
# init vars
super().__init__()
- self.search_query = search_query
- self.result_container = ResultContainer()
- self.start_time = None
- self.actual_timeout = None
-
- def search_external_bang(self):
- """
- Check if there is a external bang.
- If yes, update self.result_container and return True
- """
+ self.search_query: "SearchQuery" = search_query
+ self.result_container: ResultContainer = ResultContainer()
+ self.start_time: float | None = None
+ self.actual_timeout: float | None = None
+
+ def search_external_bang(self) -> bool:
+ """Check if there is a external bang. If yes, update
+ self.result_container and return True."""
if self.search_query.external_bang:
self.result_container.redirect_url = get_bang_url(self.search_query)
- # This means there was a valid bang and the
- # rest of the search does not need to be continued
+ # This means there was a valid bang and the rest of the search does
+ # not need to be continued
if isinstance(self.result_container.redirect_url, str):
return True
return False
@@ -72,13 +78,13 @@ class Search:
def search_answerers(self):
results = searx.answerers.STORAGE.ask(self.search_query.query)
- self.result_container.extend(None, results)
+ self.result_container.extend(None, results) # pyright: ignore[reportArgumentType]
return bool(results)
# do search-request
- def _get_requests(self):
+ def _get_requests(self) -> tuple[list[tuple[str, str, dict[str, t.Any]]], int]:
# init vars
- requests = []
+ requests: list[tuple[str, str, dict[str, t.Any]]] = []
# max of all selected engine timeout
default_timeout = 0
@@ -130,7 +136,7 @@ class Search:
return requests, actual_timeout
- def search_multiple_requests(self, requests):
+ def search_multiple_requests(self, requests: list[tuple[str, str, dict[str, t.Any]]]):
# pylint: disable=protected-access
search_id = str(uuid4())
@@ -181,7 +187,7 @@ class SearchWithPlugins(Search):
__slots__ = 'user_plugins', 'request'
- def __init__(self, search_query: SearchQuery, request: SXNG_Request, user_plugins: list[str]):
+ def __init__(self, search_query: "SearchQuery", request: "SXNG_Request", user_plugins: list[str]):
super().__init__(search_query)
self.user_plugins = user_plugins
self.result_container.on_result = self._on_result
diff --git a/searx/search/processors/__init__.py b/searx/search/processors/__init__.py
index ea049e79f..760513253 100644
--- a/searx/search/processors/__init__.py
+++ b/searx/search/processors/__init__.py
@@ -1,8 +1,5 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
-
-"""Implement request processors used by engine-types.
-
-"""
+"""Implement request processors used by engine-types."""
__all__ = [
'EngineProcessor',
@@ -14,8 +11,9 @@ __all__ = [
'PROCESSORS',
]
+import typing as t
+
import threading
-from typing import Dict
from searx import logger
from searx import engines
@@ -27,15 +25,18 @@ from .online_currency import OnlineCurrencyProcessor
from .online_url_search import OnlineUrlSearchProcessor
from .abstract import EngineProcessor
+if t.TYPE_CHECKING:
+ from searx.enginelib import Engine
+
logger = logger.getChild('search.processors')
-PROCESSORS: Dict[str, EngineProcessor] = {}
+PROCESSORS: dict[str, EngineProcessor] = {}
"""Cache request processors, stored by *engine-name* (:py:func:`initialize`)
:meta hide-value:
"""
-def get_processor_class(engine_type):
+def get_processor_class(engine_type: str) -> type[EngineProcessor] | None:
"""Return processor class according to the ``engine_type``"""
for c in [
OnlineProcessor,
@@ -49,34 +50,35 @@ def get_processor_class(engine_type):
return None
-def get_processor(engine, engine_name):
- """Return processor instance that fits to ``engine.engine.type``)"""
+def get_processor(engine: "Engine | ModuleType", engine_name: str) -> EngineProcessor | None:
+ """Return processor instance that fits to ``engine.engine.type``"""
engine_type = getattr(engine, 'engine_type', 'online')
processor_class = get_processor_class(engine_type)
- if processor_class:
+ if processor_class is not None:
return processor_class(engine, engine_name)
return None
-def initialize_processor(processor):
+def initialize_processor(processor: EngineProcessor):
"""Initialize one processor
Call the init function of the engine
"""
if processor.has_initialize_function:
- t = threading.Thread(target=processor.initialize, daemon=True)
- t.start()
+ _t = threading.Thread(target=processor.initialize, daemon=True)
+ _t.start()
-def initialize(engine_list):
- """Initialize all engines and store a processor for each engine in :py:obj:`PROCESSORS`."""
+def initialize(engine_list: list[dict[str, t.Any]]):
+ """Initialize all engines and store a processor for each engine in
+ :py:obj:`PROCESSORS`."""
for engine_data in engine_list:
- engine_name = engine_data['name']
+ engine_name: str = engine_data['name']
engine = engines.engines.get(engine_name)
if engine:
processor = get_processor(engine, engine_name)
- initialize_processor(processor)
if processor is None:
engine.logger.error('Error get processor for engine %s', engine_name)
else:
+ initialize_processor(processor)
PROCESSORS[engine_name] = processor
diff --git a/searx/search/processors/abstract.py b/searx/search/processors/abstract.py
index ff6c8f78a..2dd56855a 100644
--- a/searx/search/processors/abstract.py
+++ b/searx/search/processors/abstract.py
@@ -3,10 +3,12 @@
"""
+import typing as t
+
+import logging
import threading
from abc import abstractmethod, ABC
from timeit import default_timer
-from typing import Dict, Union
from searx import settings, logger
from searx.engines import engines
@@ -15,8 +17,11 @@ from searx.metrics import histogram_observe, counter_inc, count_exception, count
from searx.exceptions import SearxEngineAccessDeniedException, SearxEngineResponseException
from searx.utils import get_engine_from_settings
+if t.TYPE_CHECKING:
+ from searx.enginelib import Engine
+
logger = logger.getChild('searx.search.processor')
-SUSPENDED_STATUS: Dict[Union[int, str], 'SuspendedStatus'] = {}
+SUSPENDED_STATUS: dict[int | str, 'SuspendedStatus'] = {}
class SuspendedStatus:
@@ -25,16 +30,16 @@ class SuspendedStatus:
__slots__ = 'suspend_end_time', 'suspend_reason', 'continuous_errors', 'lock'
def __init__(self):
- self.lock = threading.Lock()
- self.continuous_errors = 0
- self.suspend_end_time = 0
- self.suspend_reason = None
+ self.lock: threading.Lock = threading.Lock()
+ self.continuous_errors: int = 0
+ self.suspend_end_time: float = 0
+ self.suspend_reason: str = ""
@property
def is_suspended(self):
return self.suspend_end_time >= default_timer()
- def suspend(self, suspended_time, suspend_reason):
+ def suspend(self, suspended_time: int, suspend_reason: str):
with self.lock:
# update continuous_errors / suspend_end_time
self.continuous_errors += 1
@@ -52,21 +57,21 @@ class SuspendedStatus:
# reset the suspend variables
self.continuous_errors = 0
self.suspend_end_time = 0
- self.suspend_reason = None
+ self.suspend_reason = ""
class EngineProcessor(ABC):
"""Base classes used for all types of request processors."""
- __slots__ = 'engine', 'engine_name', 'lock', 'suspended_status', 'logger'
+ __slots__ = 'engine', 'engine_name', 'suspended_status', 'logger'
- def __init__(self, engine, engine_name: str):
- self.engine = engine
- self.engine_name = engine_name
- self.logger = engines[engine_name].logger
+ def __init__(self, engine: "Engine|ModuleType", engine_name: str):
+ self.engine: "Engine" = engine
+ self.engine_name: str = engine_name
+ self.logger: logging.Logger = engines[engine_name].logger
key = get_network(self.engine_name)
key = id(key) if key else self.engine_name
- self.suspended_status = SUSPENDED_STATUS.setdefault(key, SuspendedStatus())
+ self.suspended_status: SuspendedStatus = SUSPENDED_STATUS.setdefault(key, SuspendedStatus())
def initialize(self):
try:
@@ -135,7 +140,7 @@ class EngineProcessor(ABC):
return True
return False
- def get_params(self, search_query, engine_category):
+ def get_params(self, search_query, engine_category) -> dict[str, t.Any]:
"""Returns a set of (see :ref:`request params <engine request arguments>`) or
``None`` if request is not supported.