summaryrefslogtreecommitdiff
path: root/searx/results.py
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/results.py
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/results.py')
-rw-r--r--searx/results.py35
1 files changed, 20 insertions, 15 deletions
diff --git a/searx/results.py b/searx/results.py
index 8173d85fa..02b7bb204 100644
--- a/searx/results.py
+++ b/searx/results.py
@@ -1,11 +1,11 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# pylint: disable=missing-module-docstring, missing-class-docstring
-from __future__ import annotations
+
+import typing as t
import warnings
from collections import defaultdict
from threading import RLock
-from typing import List, NamedTuple, Set
from searx import logger as log
import searx.engines
@@ -14,7 +14,10 @@ from searx.result_types import Result, LegacyResult, MainResult
from searx.result_types.answer import AnswerSet, BaseAnswer
-def calculate_score(result, priority) -> float:
+def calculate_score(
+ result: MainResult | LegacyResult,
+ priority: MainResult.PriorityType,
+) -> float:
weight = 1.0
for result_engine in result['engines']:
@@ -35,13 +38,13 @@ def calculate_score(result, priority) -> float:
return score
-class Timing(NamedTuple):
+class Timing(t.NamedTuple):
engine: str
total: float
load: float
-class UnresponsiveEngine(NamedTuple):
+class UnresponsiveEngine(t.NamedTuple):
engine: str
error_type: str
suspended: bool
@@ -70,14 +73,16 @@ class ResultContainer:
self.engine_data: dict[str, dict[str, str]] = defaultdict(dict)
self._closed: bool = False
self.paging: bool = False
- self.unresponsive_engines: Set[UnresponsiveEngine] = set()
- self.timings: List[Timing] = []
+ self.unresponsive_engines: set[UnresponsiveEngine] = set()
+ self.timings: list[Timing] = []
self.redirect_url: str | None = None
- self.on_result = lambda _: True
- self._lock = RLock()
+ self.on_result: t.Callable[[Result | LegacyResult], bool] = lambda _: True
+ self._lock: RLock = RLock()
self._main_results_sorted: list[MainResult | LegacyResult] = None # type: ignore
- def extend(self, engine_name: str | None, results): # pylint: disable=too-many-branches
+ def extend(
+ self, engine_name: str | None, results: list[Result | LegacyResult]
+ ): # pylint: disable=too-many-branches
if self._closed:
log.debug("container is closed, ignoring results: %s", results)
return
@@ -165,7 +170,7 @@ class ResultContainer:
if add_infobox:
self.infoboxes.append(new_infobox)
- def _merge_main_result(self, result: MainResult | LegacyResult, position):
+ def _merge_main_result(self, result: MainResult | LegacyResult, position: int):
result_hash = hash(result)
with self._lock:
@@ -203,8 +208,8 @@ class ResultContainer:
results = sorted(self.main_results_map.values(), key=lambda x: x.score, reverse=True)
# pass 2 : group results by category and template
- gresults = []
- categoryPositions = {}
+ gresults: list[MainResult | LegacyResult] = []
+ categoryPositions: dict[str, t.Any] = {}
max_count = 8
max_distance = 20
@@ -281,7 +286,7 @@ class ResultContainer:
return
self.timings.append(Timing(engine_name, total=engine_time, load=page_load_time))
- def get_timings(self):
+ def get_timings(self) -> list[Timing]:
with self._lock:
if not self._closed:
log.error("call to ResultContainer.get_timings before ResultContainer.close")
@@ -328,7 +333,7 @@ def merge_two_infoboxes(origin: LegacyResult, other: LegacyResult):
if not origin.attributes:
origin.attributes = other.attributes
else:
- attr_names_1 = set()
+ attr_names_1: set[str] = set()
for attr in origin.attributes:
label = attr.get("label")
if label: