client.py

"""
Single-file lightweight Python client for APINEED Google Search API.
Requires: pip install requests
"""

import json
import os
from typing import Any, Dict, Generator, List, Optional, Union
import requests


class ApineedError(Exception):
    """Base exception for APINEED API errors."""
    pass


class GoogleSearchClient:
    """Google Search API Client powered by APINEED."""

    def __init__(self, api_key: Optional[str] = None, base_url: str = "https://apineed.com"):
        self.api_key = api_key or os.getenv("APINEED_API_KEY")
        if not self.api_key:
            raise ValueError(
                "APINEED API Key is required. Pass it via GoogleSearchClient(api_key='...') "
                "or set the APINEED_API_KEY environment variable. "
                "Get one at https://apineed.com/tools/google-search"
            )
        self.base_url = base_url.rstrip("/")
        self._session = requests.Session()
        self._session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "User-Agent": "GoogleSearchClient/1.0",
        })


    def query(
        self,
        query: str,
        country: Optional[str] = None,
        language: Optional[str] = "en",
        time_range: Optional[str] = None,
        max_results: Optional[int] = 10,
        page: Optional[int] = 1,
        autocorrect: Optional[bool] = True,
    ) -> Dict[str, Any]:
        """Search Google SERP and return normalized results."""
        payload = {}
        payload["query"] = query
        if country is not None:
            payload["country"] = country
        if language is not None:
            payload["language"] = language
        if time_range is not None:
            payload["time_range"] = time_range
        if max_results is not None:
            payload["max_results"] = max_results
        if page is not None:
            payload["page"] = page
        if autocorrect is not None:
            payload["autocorrect"] = autocorrect

        res = self._session.post(
            f"{self.base_url}/v1/tools/google_search/invoke",
            json=payload,
        )
        if not res.ok:
            raise ApineedError(
                f"query failed [{res.status_code}]: {res.text}"
            )
        return res.json().get("output", {})