48 lines
1.3 KiB
Python
48 lines
1.3 KiB
Python
import os
|
|
import httpx
|
|
from typing import Optional, Dict, Any
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
GOOGLE_CSE_ENDPOINT = "https://www.googleapis.com/customsearch/v1"
|
|
|
|
|
|
class CustomSearchService:
|
|
def __init__(
|
|
self,
|
|
api_key: Optional[str] = None,
|
|
cx: Optional[str] = None,
|
|
timeout: float = 30.0,
|
|
):
|
|
self.api_key = api_key or os.getenv("GEMINI_API_KEY")
|
|
self.cx = cx or os.getenv("SEARCH_ENGINE_ID")
|
|
self.timeout = timeout
|
|
|
|
if not self.api_key:
|
|
raise ValueError("GEMINI_API_KEY is not set in environment")
|
|
if not self.cx:
|
|
raise ValueError("SEARCH_ENGINE_ID is not set in environment")
|
|
|
|
async def search(
|
|
self,
|
|
query: str,
|
|
num: int = 10,
|
|
start: int = 1,
|
|
extra_params: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
params: Dict[str, Any] = {
|
|
"key": self.api_key,
|
|
"cx": self.cx,
|
|
"q": query,
|
|
"num": num,
|
|
"start": start,
|
|
}
|
|
if extra_params:
|
|
params.update(extra_params)
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(GOOGLE_CSE_ENDPOINT, params=params)
|
|
response.raise_for_status()
|
|
return response.json()
|