Coverage for open_webui/retrieval/web/searchapi.py: 28%

1import logging 

2from typing import Optional 

3from urllib.parse import urlencode 

4 

5import requests 

6from open_webui.retrieval.web.main import SearchResult, get_filtered_results 

7 

8log = logging.getLogger(__name__) 

9 

10 

11def search_searchapi( 

12 api_key: str, 

13 engine: str, 

14 query: str, 

15 count: int, 

16 filter_list: Optional[list[str]] = None, 

17) -> list[SearchResult]: 

18 """Search using searchapi.io's API and return the results as a list of SearchResult objects. 

19 

20 Args: 

21 api_key (str): A searchapi.io API key 

22 query (str): The query to search for 

23 """ 

24 url = 'https://www.searchapi.io/api/v1/search' 

25 

26 engine = engine or 'google' 

27 

28 payload = {'engine': engine, 'q': query, 'api_key': api_key} 

29 if engine.startswith('google'): 

30 payload['link'] = 'resolved' 

31 

32 url = f'{url}?{urlencode(payload)}' 

33 response = requests.request('GET', url, timeout=30) 

34 response.raise_for_status() 

35 

36 json_response = response.json() 

37 log.debug('results from searchapi search: %s', json_response) 

38 

39 # top_stories entries carry no position, so the merged list keeps API order 

40 results = [ 

41 *json_response.get('organic_results', []), 

42 *json_response.get('top_stories', []), 

43 ] 

44 if filter_list: 

45 results = get_filtered_results(results, filter_list) 

46 return [ 

47 SearchResult( 

48 link=result.get('link', ''), 

49 title=result.get('title'), 

50 snippet=result.get('snippet'), 

51 ) 

52 for result in results[:count] 

53 ]