searx.search.Search Class Reference

Inheritance diagram for searx.search.Search:

Collaboration diagram for searx.search.Search:

Public Member Functions
	__init__ (self, "SearchQuery" search_query)
bool	search_external_bang (self)
	search_answerers (self)
	search_multiple_requests (self, list[tuple[str, str, dict[str, t.Any]]] requests)
	search_standard (self)
ResultContainer	search (self)

Public Attributes
	search_query = search_query
ResultContainer	result_container = ResultContainer()
float|None	start_time = None
float|None	actual_timeout = None

Protected Member Functions
tuple[list[tuple[str, str, dict[str, t.Any]]], int]	_get_requests (self)

Detailed Description

Search information container

Definition at line 51 of file __init__.py.

Constructor & Destructor Documentation

__init__()

searx.search.Search.__init__	(		self,
		"SearchQuery"	search_query )

Initialize the Search

Definition at line 54 of file __init__.py.

    def __init__(self, search_query: "SearchQuery"):
        """Initialize the Search"""
        # init vars
        super().__init__()
        self.search_query: "SearchQuery" = search_query
        self.result_container: ResultContainer = ResultContainer()
        self.start_time: float | None = None
        self.actual_timeout: float | None = None
 

Member Function Documentation

_get_requests()

tuple[list[tuple[str, str, dict[str, t.Any]]], int] searx.search.Search._get_requests ( self )

protected

Definition at line 82 of file __init__.py.

    def _get_requests(self) -> tuple[list[tuple[str, str, dict[str, t.Any]]], int]:
        # init vars
        requests: list[tuple[str, str, dict[str, t.Any]]] = []
 
        # max of all selected engine timeout
        default_timeout = 0
 
        # start search-request for all selected engines
        for engineref in self.search_query.engineref_list:
            processor = PROCESSORS[engineref.name]
 
            # stop the request now if the engine is suspend
            if processor.extend_container_if_suspended(self.result_container):
                continue
 
            # set default request parameters
            request_params = processor.get_params(self.search_query, engineref.category)
            if request_params is None:
                continue
 
            counter_inc('engine', engineref.name, 'search', 'count', 'sent')
 
            # append request to list
            requests.append((engineref.name, self.search_query.query, request_params))
 
            # update default_timeout
            default_timeout = max(default_timeout, processor.engine.timeout)
 
        # adjust timeout
        max_request_timeout = settings['outgoing']['max_request_timeout']
        actual_timeout = default_timeout
        query_timeout = self.search_query.timeout_limit
 
        if max_request_timeout is None and query_timeout is None:
            # No max, no user query: default_timeout
            pass
        elif max_request_timeout is None and query_timeout is not None:
            # No max, but user query: From user query except if above default
            actual_timeout = min(default_timeout, query_timeout)
        elif max_request_timeout is not None and query_timeout is None:
            # Max, no user query: Default except if above max
            actual_timeout = min(default_timeout, max_request_timeout)
        elif max_request_timeout is not None and query_timeout is not None:
            # Max & user query: From user query except if above max
            actual_timeout = min(query_timeout, max_request_timeout)
 
        logger.debug(
            "actual_timeout={0} (default_timeout={1}, ?timeout_limit={2}, max_request_timeout={3})".format(
                actual_timeout, default_timeout, query_timeout, max_request_timeout
            )
        )
 
        return requests, actual_timeout
 

search()

ResultContainer searx.search.Search.search

(

self

)

Reimplemented in searx.search.SearchWithPlugins.

Definition at line 174 of file __init__.py.

    def search(self) -> ResultContainer:
        self.start_time = default_timer()
        if not self.search_external_bang():
            if not self.search_answerers():
                self.search_standard()
        return self.result_container
 
 

search_answerers()

searx.search.Search.search_answerers

(

self

)

Definition at line 75 of file __init__.py.

    def search_answerers(self):
 
        results = searx.answerers.STORAGE.ask(self.search_query.query)
        self.result_container.extend(None, results)  # pyright: ignore[reportArgumentType]
        return bool(results)
 

search_external_bang()

bool searx.search.Search.search_external_bang

(

self

)

Check if there is a external bang.  If yes, update
self.result_container and return True.

Definition at line 63 of file __init__.py.

    def search_external_bang(self) -> bool:
        """Check if there is a external bang.  If yes, update
        self.result_container and return True."""
        if self.search_query.external_bang:
            self.result_container.redirect_url = get_bang_url(self.search_query)
 
            # This means there was a valid bang and the rest of the search does
            # not need to be continued
            if isinstance(self.result_container.redirect_url, str):
                return True
        return False
 

search_multiple_requests()

searx.search.Search.search_multiple_requests	(		self,
		list[tuple[str, str, dict[str, t.Any]]]	requests )

Definition at line 136 of file __init__.py.

    def search_multiple_requests(self, requests: list[tuple[str, str, dict[str, t.Any]]]):
        # pylint: disable=protected-access
        search_id = str(uuid4())
 
        for engine_name, query, request_params in requests:
            _search = copy_current_request_context(PROCESSORS[engine_name].search)
            th = threading.Thread(  # pylint: disable=invalid-name
                target=_search,
                args=(query, request_params, self.result_container, self.start_time, self.actual_timeout),
                name=search_id,
            )
            th._timeout = False
            th._engine_name = engine_name
            th.start()
 
        for th in threading.enumerate():  # pylint: disable=invalid-name
            if th.name == search_id:
                remaining_time = max(0.0, self.actual_timeout - (default_timer() - self.start_time))
                th.join(remaining_time)
                if th.is_alive():
                    th._timeout = True
                    self.result_container.add_unresponsive_engine(th._engine_name, 'timeout')
                    PROCESSORS[th._engine_name].logger.error('engine timeout')
 

search_standard()

searx.search.Search.search_standard

(

self

)

Update self.result_container, self.actual_timeout

Definition at line 160 of file __init__.py.

    def search_standard(self):
        """
        Update self.result_container, self.actual_timeout
        """
        requests, self.actual_timeout = self._get_requests()
 
        # send all search-request
        if requests:
            self.search_multiple_requests(requests)
 
        # return results, suggestions, answers and infoboxes
        return True
 

Member Data Documentation

actual_timeout

float | None searx.search.Search.actual_timeout = None

Definition at line 61 of file __init__.py.

result_container

searx.search.Search.result_container = ResultContainer()

Definition at line 59 of file __init__.py.

Referenced by searx.search.checker.impl.ResultContainerTests.check_basic(), searx.search.checker.impl.ResultContainerTests.has_infobox(), and searx.search.checker.impl.ResultContainerTests.result_urls().

search_query

searx.search.Search.search_query = search_query

Definition at line 58 of file __init__.py.

Referenced by searx.search.checker.impl.ResultContainerTests._record_error().

start_time

float | None searx.search.Search.start_time = None

Definition at line 60 of file __init__.py.

---

The documentation for this class was generated from the following file:

• /home/andrew/Documents/code/public/searxng/searx/search/__init__.py