summaryrefslogtreecommitdiff
path: root/searx/search/processors/__init__.py
diff options
context:
space:
mode:
Diffstat (limited to 'searx/search/processors/__init__.py')
-rw-r--r--searx/search/processors/__init__.py36
1 files changed, 19 insertions, 17 deletions
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