from fastapi import FastAPI, Query from pytrends.request import TrendReq from fastapi.responses import JSONResponse from typing import List, Optional import pandas as pd import json import time from functools import lru_cache app = FastAPI( title="PyTrends API", description="HTTP API for Google Trends data", version="1.0.0" ) # Initialize pytrends pytrends = TrendReq(hl="en-US", tz=360) # Helper function for retry logic def retry_with_backoff(func, max_retries=3, initial_delay=1, backoff_factor=2): """Retry function with exponential backoff for handling rate limits""" for attempt in range(max_retries): try: return func() except Exception as e: error_str = str(e) # Check if it's a rate limit error (429) if "429" in error_str or "rate" in error_str.lower(): if attempt < max_retries - 1: wait_time = initial_delay * (backoff_factor ** attempt) print(f"Rate limited. Waiting {wait_time} seconds before retry...") time.sleep(wait_time) continue else: return {"error": f"Service temporarily unavailable. Please try again in a few moments. Details: {error_str}"} else: raise return {"error": "Max retries exceeded"} @app.get("/") def root(): return { "message": "PyTrends API", "endpoints": { "/health": "Health check endpoint", "/interest_over_time": "Get interest over time for keywords", "/interest_by_region": "Get interest by region for keywords", "/trending_searches": "Get trending searches for a country", "/related_queries": "Get related queries for keywords" }, "note": "If you get a 429 error, Google Trends is rate limiting. Wait a few minutes and try again." } @app.get("/health") def health_check(): return {"status": "healthy"} @app.get("/interest_over_time") def get_interest_over_time( kw: List[str] = Query(..., description="Keywords to search"), timeframe: str = Query("today 5-y", description="Timeframe for search"), geo: str = Query("", description="Geographic location"), gprop: str = Query("", description="Google property (images, news, youtube, etc.)"), cat: int = Query(0, description="Category") ): try: def fetch_data(): pytrends.build_payload( kw_list=kw, timeframe=timeframe, geo=geo, gprop=gprop, cat=cat ) time.sleep(0.5) # Small delay to avoid rate limits df = pytrends.interest_over_time() if df is None or df.empty: return {"error": "No data available for this query"} df = df.drop('isPartial', axis=1) data = df.reset_index().to_dict('records') for record in data: if 'date' in record: record['date'] = str(record['date']) return { "data": data, "keywords": kw, "timeframe": timeframe, "geo": geo } result = retry_with_backoff(fetch_data) if isinstance(result, dict) and "error" in result: return JSONResponse(status_code=429, content=result) return result except Exception as e: error_msg = str(e) if "429" in error_msg or "rate" in error_msg.lower(): return JSONResponse( status_code=429, content={"error": "Too many requests to Google Trends. Please wait a few minutes before trying again."} ) return JSONResponse( status_code=400, content={"error": error_msg} ) @app.get("/interest_by_region") def get_interest_by_region( kw: List[str] = Query(..., description="Keywords to search"), timeframe: str = Query("today 5-y", description="Timeframe for search"), geo: str = Query("", description="Geographic location"), gprop: str = Query("", description="Google property"), resolution: str = Query("country", description="Resolution (country, region, metro, city)") ): try: def fetch_data(): pytrends.build_payload( kw_list=kw, timeframe=timeframe, geo=geo, gprop=gprop ) time.sleep(0.5) df = pytrends.interest_by_region(resolution=resolution) if df is None or df.empty: return {"error": "No regional data available"} data = df.reset_index().to_dict('records') return { "data": data, "keywords": kw, "resolution": resolution } result = retry_with_backoff(fetch_data) if isinstance(result, dict) and "error" in result: return JSONResponse(status_code=429, content=result) return result except Exception as e: error_msg = str(e) if "429" in error_msg or "rate" in error_msg.lower(): return JSONResponse( status_code=429, content={"error": "Too many requests to Google Trends. Please wait a few minutes before trying again."} ) return JSONResponse( status_code=400, content={"error": error_msg} ) @app.get("/trending_searches") def get_trending_searches( country: str = Query("united_states", description="Country code") ): try: def fetch_data(): time.sleep(0.5) df = pytrends.trending_searches(pn=country) if df is None or df.empty: return {"error": "No trending data available for this country"} data = df.values.tolist() return { "trending": data, "country": country } result = retry_with_backoff(fetch_data) if isinstance(result, dict) and "error" in result: return JSONResponse(status_code=429, content=result) return result except Exception as e: error_msg = str(e) if "429" in error_msg or "rate" in error_msg.lower(): return JSONResponse( status_code=429, content={"error": "Too many requests to Google Trends. Please wait a few minutes before trying again."} ) return JSONResponse( status_code=400, content={"error": error_msg} ) @app.get("/related_queries") def get_related_queries( kw: List[str] = Query(..., description="Keywords to search"), timeframe: str = Query("today 5-y", description="Timeframe for search"), geo: str = Query("", description="Geographic location") ): try: def fetch_data(): pytrends.build_payload( kw_list=kw, timeframe=timeframe, geo=geo ) time.sleep(0.5) related = pytrends.related_queries() result = {} for kw_item in kw: if kw_item in related: top_queries = related[kw_item]['top'] rising_queries = related[kw_item]['rising'] result[kw_item] = { "top": top_queries.reset_index().to_dict('records') if top_queries is not None else [], "rising": rising_queries.reset_index().to_dict('records') if rising_queries is not None else [] } return { "related_queries": result, "keywords": kw } result = retry_with_backoff(fetch_data) if isinstance(result, dict) and "error" in result: return JSONResponse(status_code=429, content=result) return result except Exception as e: error_msg = str(e) if "429" in error_msg or "rate" in error_msg.lower(): return JSONResponse( status_code=429, content={"error": "Too many requests to Google Trends. Please wait a few minutes before trying again."} ) return JSONResponse( status_code=400, content={"error": error_msg} )