Coverage for open_webui/retrieval/web/serper.py: 37%

1from __future__ import annotations 

2 

3import logging 

4 

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

6from open_webui.utils.json_codec import JSONCodec 

7from open_webui.utils.session_pool import get_session 

8 

9log = logging.getLogger(__name__) 

10 

11 

12async def search_serper( 

13 api_key: str, 

14 query: str, 

15 count: int, 

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

17) -> list[SearchResult]: 

18 """Query the serper.dev Google Search API and return normalised results. 

19 

20 Results are sorted by their position field before truncation. 

21 """ 

22 url = 'https://google.serper.dev/search' 

23 headers = {'X-API-KEY': api_key, 'Content-Type': 'application/json'} 

24 

25 session = await get_session() 

26 async with session.post(url, headers=headers, data=JSONCodec.dumps({'q': query})) as response: 

27 response.raise_for_status() 

28 payload = await response.json() 

29 

30 organic = sorted(payload.get('organic', []), key=lambda item: item.get('position', 0)) 

31 if filter_list: 

32 organic = get_filtered_results(organic, filter_list) 

33 

34 return [ 

35 SearchResult( 

36 link=item.get('link', ''), 

37 title=item.get('title'), 

38 snippet=item.get('snippet'), 

39 ) 

40 for item in organic[:count] 

41 ]