Coverage for open_webui/retrieval/web/kagi.py: 30%

1import logging 

2from typing import Optional 

3 

4import requests 

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

6 

7log = logging.getLogger(__name__) 

8 

9 

10def search_kagi(api_key: str, query: str, count: int, filter_list: Optional[list[str]] = None) -> list[SearchResult]: 

11 """Search using Kagi's Search API and return the results as a list of SearchResult objects. 

12 

13 The Search API will inherit the settings in your account, including results personalization and snippet length. 

14 

15 Args: 

16 api_key (str): A Kagi Search API key 

17 query (str): The query to search for 

18 count (int): The number of results to return 

19 """ 

20 url = 'https://kagi.com/api/v1/search' 

21 headers = { 

22 'Authorization': f'Bearer {api_key}', 

23 } 

24 params = {'query': query, 'limit': count} 

25 

26 response = requests.post(url, headers=headers, json=params) 

27 response.raise_for_status() 

28 json_response = response.json() 

29 search_results = json_response.get('data', {}).get('search', []) 

30 

31 results = [ 

32 SearchResult(link=result['url'], title=result['title'], snippet=result.get('snippet')) 

33 for result in search_results 

34 ] 

35 

36 print(results) 

37 

38 if filter_list: 

39 results = get_filtered_results(results, filter_list) 

40 

41 return results