Implement models for searx/answerers

sort froms

Import List from typing to support Python 3.8

Use SearchQuery model

Remove list for List

use Dict instead of dict

Use RawTextQuery instead of SearchQuery, type a dict, and remove unecessary str() method in webapp

improve docstring, remove test code

Implement a BaseQuery class and use that, improve answerer tests based on updated types

Add back sys

fix new linting issues

add space

Update answerer.py - use dict

use future annotations

use BaseQuery for RawTextQuery

.python-version

@@ -0,0 +1 @@
+3.8.0

searx/answerers/__init__.py

@@ -1,17 +1,20 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 # pylint: disable=missing-module-docstring
 
+from __future__ import annotations
 import sys
+from collections import defaultdict
 from os import listdir
 from os.path import realpath, dirname, join, isdir
-from collections import defaultdict
+from typing import Callable
-
+from searx.answerers.models import AnswerModule, AnswerDict
+from searx.search.models import BaseQuery
 from searx.utils import load_module
 
 answerers_dir = dirname(realpath(__file__))
 
 
-def load_answerers():
+def load_answerers() -> list[AnswerModule]:
     answerers = []  # pylint: disable=redefined-outer-name
 
     for filename in listdir(answerers_dir):
@@ -24,7 +27,9 @@ def load_answerers():
     return answerers
 
 
-def get_answerers_by_keywords(answerers):  # pylint:disable=redefined-outer-name
+def get_answerers_by_keywords(
+    answerers: list[AnswerModule],  # pylint: disable=redefined-outer-name
+) -> dict[str, list[Callable[[BaseQuery], list[AnswerDict]]]]:
     by_keyword = defaultdict(list)
     for answerer in answerers:
         for keyword in answerer.keywords:
@@ -33,8 +38,8 @@ def get_answerers_by_keywords(answerers):  # pylint:disable=redefined-outer-name
     return by_keyword
 
 
-def ask(query):
+def ask(query: BaseQuery) -> list[list[AnswerDict]]:
-    results = []
+    results: list[list[AnswerDict]] = []
     query_parts = list(filter(None, query.query.split()))
 
     if not query_parts or query_parts[0] not in answerers_by_keywords:

searx/answerers/models.py

@@ -0,0 +1,38 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+# pylint: disable=missing-module-docstring
+
+from __future__ import annotations
+from typing import TypedDict, Tuple
+from abc import abstractmethod, ABC
+from searx.search.models import BaseQuery
+
+
+class AnswerDict(TypedDict):
+    """The result of a given answer response"""
+
+    answer: str
+
+
+class AnswerSelfInfoDict(TypedDict):
+    """The information about the AnswerModule"""
+
+    name: str
+    description: str
+    examples: list[str]
+
+
+class AnswerModule(ABC):
+    """A module which returns possible answers for auto-complete requests"""
+
+    @property
+    @abstractmethod
+    def keywords(self) -> Tuple[str]:
+        """Keywords which will be used to determine if the answer should be called"""
+
+    @abstractmethod
+    def answer(self, query: BaseQuery) -> list[AnswerDict]:
+        """From a query, get the possible auto-complete answers"""
+
+    @abstractmethod
+    def self_info(self) -> AnswerSelfInfoDict:
+        """Provides information about the AnswerModule"""

searx/answerers/random/answerer.py

@@ -1,10 +1,14 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
+from __future__ import annotations
 import hashlib
 import random
 import string
 import uuid
 from flask_babel import gettext
+from typing import Callable
+from searx.answerers.models import AnswerDict, AnswerSelfInfoDict
+from searx.search.models import BaseQuery
 
 # required answerer attribute
 # specifies which search query keywords triggers this answerer
@@ -45,7 +49,7 @@ def random_color():
     return f"#{color.upper()}"
 
 
-random_types = {
+random_types: dict[str, Callable[[], str]] = {
     'string': random_string,
     'int': random_int,
     'float': random_float,
@@ -57,7 +61,7 @@ random_types = {
 
 # required answerer function
 # can return a list of results (any result type) for a given query
-def answer(query):
+def answer(query: BaseQuery) -> list[AnswerDict]:
     parts = query.query.split()
     if len(parts) != 2:
         return []
@@ -70,7 +74,7 @@ def answer(query):
 
 # required answerer function
 # returns information about the answerer
-def self_info():
+def self_info() -> AnswerSelfInfoDict:
     return {
         'name': gettext('Random value generator'),
         'description': gettext('Generate different random values'),

searx/answerers/statistics/answerer.py

@@ -1,49 +1,51 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
+
+from __future__ import annotations
 from functools import reduce
 from operator import mul
 
 from flask_babel import gettext
+from typing import Callable
+from searx.answerers.models import AnswerDict, AnswerSelfInfoDict
+from searx.search.models import BaseQuery
 
 
 keywords = ('min', 'max', 'avg', 'sum', 'prod')
 
 
+stastistics_map: dict[str, Callable[[list[float]], float]] = {
+    'min': lambda args: min(args),
+    'max': lambda args: max(args),
+    'avg': lambda args: sum(args) / len(args),
+    'sum': lambda args: sum(args),
+    'prod': lambda args: reduce(mul, args, 1),
+}
+
+
 # required answerer function
 # can return a list of results (any result type) for a given query
-def answer(query):
+def answer(query: BaseQuery) -> list[AnswerDict]:
     parts = query.query.split()
 
     if len(parts) < 2:
         return []
 
     try:
-        args = list(map(float, parts[1:]))
+        args: list[float] = list(map(float, parts[1:]))
-    except:
+    except Exception:
         return []
 
     func = parts[0]
-    answer = None
 
-    if func == 'min':
+    if func not in stastistics_map:
-        answer = min(args)
-    elif func == 'max':
-        answer = max(args)
-    elif func == 'avg':
-        answer = sum(args) / len(args)
-    elif func == 'sum':
-        answer = sum(args)
-    elif func == 'prod':
-        answer = reduce(mul, args, 1)
-
-    if answer is None:
         return []
 
-    return [{'answer': str(answer)}]
+    return [{'answer': str(stastistics_map[func](args))}]
 
 
 # required answerer function
 # returns information about the answerer
-def self_info():
+def self_info() -> AnswerSelfInfoDict:
     return {
         'name': gettext('Statistics functions'),
         'description': gettext('Compute {functions} of the arguments').format(functions='/'.join(keywords)),

searx/query.py

@@ -5,6 +5,7 @@ from abc import abstractmethod, ABC
 import re
 
 from searx import settings
+from searx.search.models import BaseQuery
 from searx.sxng_locales import sxng_locales
 from searx.engines import categories, engines, engine_shortcuts
 from searx.external_bang import get_bang_definition_and_autocomplete
@@ -247,7 +248,7 @@ class FeelingLuckyParser(QueryPartParser):
         return True
 
 
-class RawTextQuery:
+class RawTextQuery(BaseQuery):
     """parse raw text query (the value from the html input)"""
 
     PARSER_CLASSES = [

searx/search/models.py

@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 # pylint: disable=missing-module-docstring
 
+from abc import ABC
 import typing
 import babel
 
@@ -24,7 +25,13 @@ class EngineRef:
         return hash((self.name, self.category))
 
 
-class SearchQuery:
+class BaseQuery(ABC):  # pylint: disable=too-few-public-methods
+    """Contains properties among all query classes"""
+
+    query: str
+
+
+class SearchQuery(BaseQuery):
     """container for all the search parameters (query, language, etc...)"""
 
     __slots__ = (

searx/webapp.py

@@ -851,7 +851,7 @@ def autocompleter():
 
     for answers in ask(raw_text_query):
         for answer in answers:
-            results.append(str(answer['answer']))
+            results.append(answer['answer'])
 
     if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
         # the suggestion request comes from the searx search form

tests/unit/test_answerers.py

@@ -12,5 +12,10 @@ class AnswererTest(SearxTestCase):  # pylint: disable=missing-class-docstring
         query = Mock()
         unicode_payload = 'árvíztűrő tükörfúrógép'
         for answerer in answerers:
-            query.query = '{} {}'.format(answerer.keywords[0], unicode_payload)
+            for keyword in answerer.keywords:
-            self.assertTrue(isinstance(answerer.answer(query), list))
+                query.query = '{} {}'.format(keyword, unicode_payload)
+                answer_dicts = answerer.answer(query)
+                self.assertTrue(isinstance(answer_dicts, list))
+                for answer_dict in answer_dicts:
+                    self.assertTrue('answer' in answer_dict)
+                    self.assertTrue(isinstance(answer_dict['answer'], str))