Coverage for open_webui/retrieval/web/serpapi.py: 35%

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_serpapi( 

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 serpapi.com's API and return the results as a list of SearchResult objects. 

19 

20 Args: 

21 api_key (str): A serpapi.com API key 

22 query (str): The query to search for 

23 """ 

24 url = 'https://serpapi.com/search' 

25 

26 engine = engine or 'google' 

27 

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

29 

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

31 response = requests.request('GET', url) 

32 

33 json_response = response.json() 

34 log.info('results from serpapi search: %s', json_response) 

35 

36 results = sorted(json_response.get('organic_results', []), key=lambda x: x.get('position', 0)) 

37 if filter_list: 

38 results = get_filtered_results(results, filter_list) 

39 return [ 

40 SearchResult( 

41 link=result['link'], 

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

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

44 ) 

45 for result in results[:count] 

46 ]