File size: 2,503 Bytes
ed10b54 de25561 ed10b54 de25561 ed10b54 de25561 ed10b54 f040e7d ed10b54 6f0edd1 cc9dd3f ed10b54 7d3243b ed10b54 7d3243b ed10b54 7d3243b ed10b54 | 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 | # main.py
from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader
from fastapi.middleware.cors import CORSMiddleware
from fastapi import Request, Query
from fastapi.templating import Jinja2Templates
from fastapi import File, UploadFile
from fastapi.responses import FileResponse
from fastapi.responses import Response
from pydantic import BaseModel
from serpapi import GoogleSearch
import os
app = FastAPI(title="SerpApi Research Backend")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Security setup
api_key_header = APIKeyHeader(name="X-API-Key")
SERPAPI_KEY = os.getenv("SERPAPI_SECRET") # Set in Hugging Face secrets
class SearchQuery(BaseModel):
query: str
engine: str = "google"
num_results: int = 10
def get_serpapi_client():
if not SERPAPI_KEY:
raise HTTPException(
status_code=500,
detail="SerpApi configuration missing"
)
return GoogleSearch({"api_key": SERPAPI_KEY})
def verify_api_key(api_key: str = Security(api_key_header)):
print("api_key: " + api_key)
if api_key != os.getenv("API_SECRET_KEY"):
raise HTTPException(
status_code=401,
detail="Invalid API key"
)
return True
templates = Jinja2Templates(directory=".")
@app.get("/")
def read_root(request: Request):
return templates.TemplateResponse("index.html", {"request": request})
@app.post("/search")
async def perform_search(
search_query: SearchQuery,
_: bool = Depends(verify_api_key)
):
try:
#client = get_serpapi_client()
params = {
"q": search_query.query,
"engine": search_query.engine,
"num": search_query.num_results,
"api_key": SERPAPI_KEY
}
# Initialize GoogleSearch with parameters
search = GoogleSearch(params)
# Fetch results as a dictionary
results = search.get_dict()
#results = client.get_dict(params)
return {
"organic_results": results.get("organic_results", []),
"related_searches": results.get("related_searches", [])
}
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Search failed: {str(e)}"
)
@app.get("/health")
async def health_check():
return {"status": "operational"}
|