Run shared provider search orchestration and return ID-bearing results. This path applies optional query normalization and default limits, then delegates API access to provider hooks with consistent logging and failure handling.
(
self, query_type: QueryType, query: str, filters: dict[str, str]
)
| 401 | raise NotImplementedError |
| 402 | |
| 403 | def _search_api( |
| 404 | self, query_type: QueryType, query: str, filters: dict[str, str] |
| 405 | ) -> Sequence[R]: |
| 406 | """Run shared provider search orchestration and return ID-bearing results. |
| 407 | |
| 408 | This path applies optional query normalization and default limits, then |
| 409 | delegates API access to provider hooks with consistent logging and |
| 410 | failure handling. |
| 411 | """ |
| 412 | if self.config["search_query_ascii"].get(): |
| 413 | query = unidecode.unidecode(query) |
| 414 | |
| 415 | limit = self.config["search_limit"].get(int) |
| 416 | params = SearchParams(query_type, query, filters, limit) |
| 417 | |
| 418 | self._log.debug("Searching for '{}' with {}", query, filters) |
| 419 | try: |
| 420 | response_data = self.get_search_response(params) |
| 421 | except Exception as e: |
| 422 | if config["raise_on_error"].get(bool): |
| 423 | raise |
| 424 | self._log.error( |
| 425 | "Error searching {.data_source}: {}", self, e, exc_info=True |
| 426 | ) |
| 427 | return () |
| 428 | |
| 429 | self._log.debug("Found {} result(s)", len(response_data)) |
| 430 | return response_data |
| 431 | |
| 432 | def _get_candidates( |
| 433 | self, query_type: QueryType, *args, **kwargs |