-
Notifications
You must be signed in to change notification settings - Fork 3k
/
google_connector.py
76 lines (58 loc) · 2.88 KB
/
google_connector.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
# Copyright (c) Microsoft. All rights reserved.
import logging
import urllib
import aiohttp
from semantic_kernel.connectors.search_engine.connector import ConnectorBase
from semantic_kernel.exceptions import ServiceInitializationError, ServiceInvalidRequestError
logger: logging.Logger = logging.getLogger(__name__)
class GoogleConnector(ConnectorBase):
"""
A search engine connector that uses the Google Custom Search API to perform a web search.
"""
_api_key: str
_search_engine_id: str
def __init__(self, api_key: str, search_engine_id: str) -> None:
self._api_key = api_key
self._search_engine_id = search_engine_id
if not self._api_key:
raise ServiceInitializationError("Google Custom Search API key cannot be null.")
if not self._search_engine_id:
raise ServiceInitializationError("Google search engine ID cannot be null.")
async def search(self, query: str, num_results: int = 1, offset: int = 0) -> list[str]:
"""
Returns the search results of the query provided by pinging the Google Custom search API.
Returns `num_results` results and ignores the first `offset`.
:param query: search query
:param num_results: the number of search results to return
:param offset: the number of search results to ignore
:return: list of search results
"""
if not query:
raise ServiceInvalidRequestError("query cannot be 'None' or empty.")
if num_results <= 0:
raise ServiceInvalidRequestError("num_results value must be greater than 0.")
if num_results > 10:
raise ServiceInvalidRequestError("num_results value must be less than or equal to 10.")
if offset < 0:
raise ServiceInvalidRequestError("offset must be greater than 0.")
logger.info(
f"Received request for google search with \
params:\nquery: {query}\nnum_results: {num_results}\noffset: {offset}"
)
_base_url = "https://www.googleapis.com/customsearch/v1"
_request_url = (
f"{_base_url}?q={urllib.parse.quote_plus(query)}"
f"&key={self._api_key}&cx={self._search_engine_id}"
f"&num={num_results}&start={offset}"
)
logger.info("Sending GET request to Google Search API.")
async with aiohttp.ClientSession() as session:
async with session.get(_request_url, raise_for_status=True) as response:
if response.status == 200:
data = await response.json()
logger.info("Request successful.")
logger.info(f"API Response: {data}")
return [x["snippet"] for x in data["items"]]
else:
logger.error(f"Request to Google Search API failed with status code: {response.status}.")
return []