[typing] add type hints for dictionaries

This commit is contained in:
Martin Fischer 2022-01-17 09:04:51 +01:00
parent 6d43cf7952
commit def62c3a47
4 changed files with 10 additions and 5 deletions

View file

@ -4,7 +4,9 @@
import asyncio import asyncio
import logging import logging
from ssl import SSLContext
import threading import threading
from typing import Any, Dict
import httpx import httpx
from httpx_socks import AsyncProxyTransport from httpx_socks import AsyncProxyTransport
@ -23,7 +25,7 @@ else:
logger = logger.getChild('searx.network.client') logger = logger.getChild('searx.network.client')
LOOP = None LOOP = None
SSLCONTEXTS = {} SSLCONTEXTS: Dict[Any, SSLContext] = {}
TRANSPORT_KWARGS = { TRANSPORT_KWARGS = {
'trust_env': False, 'trust_env': False,
} }

View file

@ -7,6 +7,7 @@ import atexit
import asyncio import asyncio
import ipaddress import ipaddress
from itertools import cycle from itertools import cycle
from typing import Dict
import httpx import httpx
@ -16,7 +17,7 @@ from .client import new_client, get_loop, AsyncHTTPTransportNoHttp
logger = logger.getChild('network') logger = logger.getChild('network')
DEFAULT_NAME = '__DEFAULT__' DEFAULT_NAME = '__DEFAULT__'
NETWORKS = {} NETWORKS: Dict[str, 'Network'] = {}
# requests compatibility when reading proxy settings from settings.yml # requests compatibility when reading proxy settings from settings.yml
PROXY_PATTERN_MAPPING = { PROXY_PATTERN_MAPPING = {
'http': 'http://', 'http': 'http://',

View file

@ -15,6 +15,7 @@ __all__ = [
] ]
import threading import threading
from typing import Dict
from searx import logger from searx import logger
from searx import engines from searx import engines
@ -26,7 +27,7 @@ from .online_currency import OnlineCurrencyProcessor
from .abstract import EngineProcessor from .abstract import EngineProcessor
logger = logger.getChild('search.processors') logger = logger.getChild('search.processors')
PROCESSORS = {} PROCESSORS: Dict[str, EngineProcessor] = {}
"""Cache request processores, stored by *engine-name* (:py:func:`initialize`)""" """Cache request processores, stored by *engine-name* (:py:func:`initialize`)"""

View file

@ -8,6 +8,7 @@
import threading import threading
from abc import abstractmethod, ABC from abc import abstractmethod, ABC
from timeit import default_timer from timeit import default_timer
from typing import Dict, Union
from searx import settings, logger from searx import settings, logger
from searx.engines import engines from searx.engines import engines
@ -17,7 +18,7 @@ from searx.exceptions import SearxEngineAccessDeniedException, SearxEngineRespon
from searx.utils import get_engine_from_settings from searx.utils import get_engine_from_settings
logger = logger.getChild('searx.search.processor') logger = logger.getChild('searx.search.processor')
SUSPENDED_STATUS = {} SUSPENDED_STATUS: Dict[Union[int, str], 'SuspendedStatus'] = {}
class SuspendedStatus: class SuspendedStatus:
@ -61,7 +62,7 @@ class EngineProcessor(ABC):
__slots__ = 'engine', 'engine_name', 'lock', 'suspended_status', 'logger' __slots__ = 'engine', 'engine_name', 'lock', 'suspended_status', 'logger'
def __init__(self, engine, engine_name): def __init__(self, engine, engine_name: str):
self.engine = engine self.engine = engine
self.engine_name = engine_name self.engine_name = engine_name
self.logger = engines[engine_name].logger self.logger = engines[engine_name].logger