searx.network Namespace Reference

Namespaces
namespace	client
namespace	network
namespace	raise_for_httperror

Classes
class	Request

Functions
	reset_time_for_thread ()
	get_time_for_thread ()
	set_timeout_for_thread (float timeout, float|None start_time=None)
	set_context_network_name (str network_name)
	get_context_network ()
	_record_http_time ()
	_get_timeout (float start_time, kwargs)
SXNG_Response	request (method, url, **kwargs)
list[httpx.Response|Exception]	multi_requests (list["Request"] request_list)
SXNG_Response	get (str url, **t.Any kwargs)
SXNG_Response	options (str url, **t.Any kwargs)
SXNG_Response	head (str url, **t.Any kwargs)
SXNG_Response	post (str url, data=None, **t.Any kwargs)
SXNG_Response	put (str url, data=None, **t.Any kwargs)
SXNG_Response	patch (str url, data=None, **t.Any kwargs)
SXNG_Response	delete (str url, **t.Any kwargs)
	stream_chunk_to_queue (network, queue, str method, str url, **t.Any kwargs)
	_stream_generator (str method, str url, **t.Any kwargs)
	_close_response_method (self)
tuple[httpx.Response, Iterable[bytes]]	stream (str method, str url, **t.Any kwargs)

Variables
list	__all__ = ["initialize", "check_network_configuration", "raise_for_httperror"]
	THREADLOCAL = threading.local()
# pylint disable	_generator = protected-access

Function Documentation

_close_response_method()

searx.network._close_response_method ( self )

protected

Definition at line 243 of file __init__.py.

def _close_response_method(self):
    asyncio.run_coroutine_threadsafe(self.aclose(), get_loop())
    # reach the end of _self.generator ( _stream_generator ) to an avoid memory leak.
    # it makes sure that :
    # * the httpx response is closed (see the stream_chunk_to_queue function)
    # * to call future.result() in _stream_generator
    for _ in self._generator:  # pylint: disable=protected-access
        continue
 
 

_get_timeout()

searx.network._get_timeout ( float start_time, kwargs )

protected

Definition at line 71 of file __init__.py.

def _get_timeout(start_time: float, kwargs):
    # pylint: disable=too-many-branches
 
    timeout: float | None
    # timeout (httpx)
    if 'timeout' in kwargs:
        timeout = kwargs['timeout']
    else:
        timeout = getattr(THREADLOCAL, 'timeout', None)
        if timeout is not None:
            kwargs['timeout'] = timeout
 
    # 2 minutes timeout for the requests without timeout
    timeout = timeout or 120
 
    # adjust actual timeout
    timeout += 0.2  # overhead
    if start_time:
        timeout -= default_timer() - start_time
 
    return timeout
 
 

Referenced by multi_requests(), and request().

Here is the caller graph for this function:

_record_http_time()

searx.network._record_http_time ( )

protected

Definition at line 57 of file __init__.py.

def _record_http_time():
    # pylint: disable=too-many-branches
    time_before_request = default_timer()
    start_time = getattr(THREADLOCAL, 'start_time', time_before_request)
    try:
        yield start_time
    finally:
        # update total_time.
        # See get_time_for_thread() and reset_time_for_thread()
        if hasattr(THREADLOCAL, 'total_time'):
            time_after_request = default_timer()
            THREADLOCAL.total_time += time_after_request - time_before_request
 
 

Referenced by multi_requests(), and request().

Here is the caller graph for this function:

_stream_generator()

searx.network._stream_generator ( str method, str url, **t.Any kwargs )

protected

Definition at line 228 of file __init__.py.

def _stream_generator(method: str, url: str, **kwargs: t.Any):
    queue = SimpleQueue()
    network = get_context_network()
    future = asyncio.run_coroutine_threadsafe(stream_chunk_to_queue(network, queue, method, url, **kwargs), get_loop())
 
    # yield chunks
    obj_or_exception = queue.get()
    while obj_or_exception is not None:
        if isinstance(obj_or_exception, Exception):
            raise obj_or_exception
        yield obj_or_exception
        obj_or_exception = queue.get()
    future.result()
 
 

References get_context_network(), and stream_chunk_to_queue().

Referenced by stream().

Here is the call graph for this function:

Here is the caller graph for this function:

delete()

SXNG_Response searx.network.delete	(	str	url,
		**t.Any	kwargs )

Definition at line 198 of file __init__.py.

def delete(url: str, **kwargs: t.Any) -> SXNG_Response:
    return request('delete', url, **kwargs)
 
 

References request().

Here is the call graph for this function:

get()

SXNG_Response searx.network.get	(	str	url,
		**t.Any	kwargs )

Definition at line 171 of file __init__.py.

def get(url: str, **kwargs: t.Any) -> SXNG_Response:
    kwargs.setdefault('allow_redirects', True)
    return request('get', url, **kwargs)
 
 

References request().

Here is the call graph for this function:

get_context_network()

searx.network.get_context_network

(

)

If set return thread's network.

If unset, return value from :py:obj:`get_network`.

Definition at line 48 of file __init__.py.

def get_context_network():
    """If set return thread's network.
 
    If unset, return value from :py:obj:`get_network`.
    """
    return THREADLOCAL.__dict__.get('network') or get_network()
 
 
@contextmanager

Referenced by _stream_generator(), multi_requests(), and request().

Here is the caller graph for this function:

get_time_for_thread()

searx.network.get_time_for_thread

(

)

returns thread's total time or None

Definition at line 34 of file __init__.py.

def get_time_for_thread():
    """returns thread's total time or None"""
    return THREADLOCAL.__dict__.get('total_time')
 
 

head()

SXNG_Response searx.network.head	(	str	url,
		**t.Any	kwargs )

Definition at line 181 of file __init__.py.

def head(url: str, **kwargs: t.Any) -> SXNG_Response:
    kwargs.setdefault('allow_redirects', False)
    return request('head', url, **kwargs)
 
 

References request().

Here is the call graph for this function:

multi_requests()

list[httpx.Response | Exception] searx.network.multi_requests

(

list["Request"]

request_list

)

send multiple HTTP requests in parallel. Wait for all requests to finish.

Definition at line 109 of file __init__.py.

def multi_requests(request_list: list["Request"]) -> list[httpx.Response | Exception]:
    """send multiple HTTP requests in parallel. Wait for all requests to finish."""
    with _record_http_time() as start_time:
        # send the requests
        network = get_context_network()
        loop = get_loop()
        future_list = []
        for request_desc in request_list:
            timeout = _get_timeout(start_time, request_desc.kwargs)
            future = asyncio.run_coroutine_threadsafe(
                network.request(request_desc.method, request_desc.url, **request_desc.kwargs), loop
            )
            future_list.append((future, timeout))
 
        # read the responses
        responses = []
        for future, timeout in future_list:
            try:
                responses.append(future.result(timeout))
            except concurrent.futures.TimeoutError:
                responses.append(httpx.TimeoutException('Timeout', request=None))
            except Exception as e:  # pylint: disable=broad-except
                responses.append(e)
        return responses
 
 

References _get_timeout(), _record_http_time(), and get_context_network().

Here is the call graph for this function:

options()

SXNG_Response searx.network.options	(	str	url,
		**t.Any	kwargs )

Definition at line 176 of file __init__.py.

def options(url: str, **kwargs: t.Any) -> SXNG_Response:
    kwargs.setdefault('allow_redirects', True)
    return request('options', url, **kwargs)
 
 

References request().

Here is the call graph for this function:

patch()

SXNG_Response searx.network.patch	(	str	url,
			data = None,
		**t.Any	kwargs )

Definition at line 194 of file __init__.py.

def patch(url: str, data=None, **kwargs: t.Any) -> SXNG_Response:
    return request('patch', url, data=data, **kwargs)
 
 

References request().

Here is the call graph for this function:

post()

SXNG_Response searx.network.post	(	str	url,
			data = None,
		**t.Any	kwargs )

Definition at line 186 of file __init__.py.

def post(url: str, data=None, **kwargs: t.Any) -> SXNG_Response:
    return request('post', url, data=data, **kwargs)
 
 

References request().

Here is the call graph for this function:

put()

SXNG_Response searx.network.put	(	str	url,
			data = None,
		**t.Any	kwargs )

Definition at line 190 of file __init__.py.

def put(url: str, data=None, **kwargs: t.Any) -> SXNG_Response:
    return request('put', url, data=data, **kwargs)
 
 

References request().

Here is the call graph for this function:

request()

SXNG_Response searx.network.request	(		method,
			url,
		**	kwargs )

same as requests/requests/api.py request(...)

Definition at line 94 of file __init__.py.

def request(method, url, **kwargs) -> SXNG_Response:
    """same as requests/requests/api.py request(...)"""
    with _record_http_time() as start_time:
        network = get_context_network()
        timeout = _get_timeout(start_time, kwargs)
        future = asyncio.run_coroutine_threadsafe(
            network.request(method, url, **kwargs),
            get_loop(),
        )
        try:
            return future.result(timeout)
        except concurrent.futures.TimeoutError as e:
            raise httpx.TimeoutException('Timeout', request=None) from e
 
 

References _get_timeout(), _record_http_time(), and get_context_network().

Referenced by delete(), get(), head(), options(), patch(), post(), and put().

Here is the call graph for this function:

Here is the caller graph for this function:

reset_time_for_thread()

searx.network.reset_time_for_thread

(

)

Definition at line 30 of file __init__.py.

def reset_time_for_thread():
    THREADLOCAL.total_time = 0
 
 

Referenced by searx.search.processors.online.OnlineProcessor.initialize(), and searx.search.processors.online.OnlineProcessor.search().

Here is the caller graph for this function:

set_context_network_name()

searx.network.set_context_network_name

(

str

network_name

)

Definition at line 44 of file __init__.py.

def set_context_network_name(network_name: str):
    THREADLOCAL.network = get_network(network_name)
 
 

Referenced by searx.search.processors.online.OnlineProcessor.search().

Here is the caller graph for this function:

set_timeout_for_thread()

searx.network.set_timeout_for_thread	(	float	timeout,
		float | None	start_time = None )

Definition at line 39 of file __init__.py.

def set_timeout_for_thread(timeout: float, start_time: float | None = None):
    THREADLOCAL.timeout = timeout
    THREADLOCAL.start_time = start_time
 
 

Referenced by searx.search.processors.online.OnlineProcessor.initialize(), and searx.search.processors.online.OnlineProcessor.search().

Here is the caller graph for this function:

stream()

tuple[httpx.Response, Iterable[bytes]] searx.network.stream	(	str	method,
		str	url,
		**t.Any	kwargs )

Replace httpx.stream.

Usage:
response, stream = poolrequests.stream(...)
for chunk in stream:
    ...

httpx.Client.stream requires to write the httpx.HTTPTransport version of the
the httpx.AsyncHTTPTransport declared above.

Definition at line 253 of file __init__.py.

def stream(method: str, url: str, **kwargs: t.Any) -> tuple[httpx.Response, Iterable[bytes]]:
    """Replace httpx.stream.
 
    Usage:
    response, stream = poolrequests.stream(...)
    for chunk in stream:
        ...
 
    httpx.Client.stream requires to write the httpx.HTTPTransport version of the
    the httpx.AsyncHTTPTransport declared above.
    """
    generator = _stream_generator(method, url, **kwargs)
 
    # yield response
    response = next(generator)  # pylint: disable=stop-iteration-return
    if isinstance(response, Exception):
        raise response
 
    response._generator = generator  # pylint: disable=protected-access
    response.close = MethodType(_close_response_method, response)
 
    return response, generator

References _stream_generator().

Here is the call graph for this function:

stream_chunk_to_queue()

searx.network.stream_chunk_to_queue	(		network,
			queue,
		str	method,
		str	url,
		**t.Any	kwargs )

Definition at line 202 of file __init__.py.

async def stream_chunk_to_queue(network, queue, method: str, url: str, **kwargs: t.Any):
    try:
        async with await network.stream(method, url, **kwargs) as response:
            queue.put(response)
            # aiter_raw: access the raw bytes on the response without applying any HTTP content decoding
            # https://www.python-httpx.org/quickstart/#streaming-responses
            async for chunk in response.aiter_raw(65536):
                if len(chunk) > 0:
                    queue.put(chunk)
    except (httpx.StreamClosed, anyio.ClosedResourceError):
        # the response was queued before the exception.
        # the exception was raised on aiter_raw.
        # we do nothing here: in the finally block, None will be queued
        # so stream(method, url, **kwargs) generator can stop
        pass
    except Exception as e:  # pylint: disable=broad-except
        # broad except to avoid this scenario:
        # exception in network.stream(method, url, **kwargs)
        # -> the exception is not catch here
        # -> queue None (in finally)
        # -> the function below steam(method, url, **kwargs) has nothing to return
        queue.put(e)
    finally:
        queue.put(None)
 
 

Referenced by _stream_generator().

Here is the caller graph for this function:

Variable Documentation

__all__

list searx.network.__all__ = ["initialize", "check_network_configuration", "raise_for_httperror"]

private

Definition at line 4 of file __init__.py.

_generator

# pylint disable searx.network._generator = protected-access

protected

Definition at line 249 of file __init__.py.

THREADLOCAL

searx.network.THREADLOCAL = threading.local()

Definition at line 26 of file __init__.py.