Spaces:
Running
Running
File size: 3,320 Bytes
d9162ac |
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 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 |
"""Serper API client for Google searches.
Vendored and adapted from folder/tools/web_search.py.
"""
import os
import aiohttp
import structlog
from src.tools.vendored.web_search_core import WebpageSnippet, ssl_context
from src.utils.exceptions import RateLimitError, SearchError
logger = structlog.get_logger()
class SerperClient:
"""A client for the Serper API to perform Google searches."""
def __init__(self, api_key: str | None = None) -> None:
"""Initialize Serper client.
Args:
api_key: Serper API key. If None, reads from SERPER_API_KEY env var.
Raises:
ConfigurationError: If no API key is provided.
"""
self.api_key = api_key or os.getenv("SERPER_API_KEY")
if not self.api_key:
from src.utils.exceptions import ConfigurationError
raise ConfigurationError(
"No API key provided. Set SERPER_API_KEY environment variable."
)
self.url = "https://google.serper.dev/search"
self.headers = {"X-API-KEY": self.api_key, "Content-Type": "application/json"}
async def search(
self, query: str, filter_for_relevance: bool = False, max_results: int = 5
) -> list[WebpageSnippet]:
"""Perform a Google search using Serper API.
Args:
query: The search query
filter_for_relevance: Whether to filter results (currently not implemented)
max_results: Maximum number of results to return
Returns:
List of WebpageSnippet objects with search results
Raises:
SearchError: If the search fails
RateLimitError: If rate limit is exceeded
"""
connector = aiohttp.TCPConnector(ssl=ssl_context)
try:
async with aiohttp.ClientSession(connector=connector) as session:
async with session.post(
self.url, headers=self.headers, json={"q": query, "autocorrect": False}
) as response:
if response.status == 429:
raise RateLimitError("Serper API rate limit exceeded")
response.raise_for_status()
results = await response.json()
results_list = [
WebpageSnippet(
url=result.get("link", ""),
title=result.get("title", ""),
description=result.get("snippet", ""),
)
for result in results.get("organic", [])
]
if not results_list:
logger.info("No search results found", query=query)
return []
# Return results up to max_results
return results_list[:max_results]
except aiohttp.ClientError as e:
logger.error("Serper API request failed", error=str(e), query=query)
raise SearchError(f"Serper API request failed: {e}") from e
except RateLimitError:
raise
except Exception as e:
logger.error("Unexpected error in Serper search", error=str(e), query=query)
raise SearchError(f"Serper search failed: {e}") from e
|