client.py

"""
Single-file lightweight Python client for APINEED Google Trends 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 GoogleTrendsClient:
    """Google Trends 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 GoogleTrendsClient(api_key='...') "
                "or set the APINEED_API_KEY environment variable. "
                "Get one at https://apineed.com/tools/google-trends"
            )
        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": "GoogleTrendsClient/1.0",
        })


    def interest_over_time(
        self,
        query: str,
        geo: Optional[str] = None,
        date: Optional[str] = "today 3-m",
        data_type: Optional[str] = "INTEREST_OVER_TIME",
        language: Optional[str] = "en",
    ) -> Dict[str, Any]:
        """Fetch search interest trend over time or compare terms."""
        payload = {}
        payload["query"] = query
        if geo is not None:
            payload["geo"] = geo
        if date is not None:
            payload["date"] = date
        if data_type is not None:
            payload["data_type"] = data_type
        if language is not None:
            payload["language"] = language

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


    def trending_now(
        self,
        geo: str = "US",
        hours: str = "PAST_24_HOURS",
        language: str = "en",
        max_results: int = 20,
        max_related_queries: int = 10,
    ) -> Dict[str, Any]:
        """Fetch current trending searches (breaking topics)."""
        payload = {}
        if geo is not None:
            payload["geo"] = geo
        if hours is not None:
            payload["hours"] = hours
        if language is not None:
            payload["language"] = language
        if max_results is not None:
            payload["max_results"] = max_results
        if max_related_queries is not None:
            payload["max_related_queries"] = max_related_queries

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


    def autocomplete(
        self,
        query: str,
        language: str = "en",
    ) -> Dict[str, Any]:
        """Fetch Google Trends search suggestions."""
        payload = {}
        payload["query"] = query
        if language is not None:
            payload["language"] = language

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