Spaces:
Runtime error
Runtime error
File size: 2,748 Bytes
ef1d021 13e5718 bf4f857 925e717 ef1d021 30a3e14 ef1d021 50981c1 13e5718 bf4f857 d1fa8b0 bf4f857 13e5718 bf4f857 34a4a57 bf4f857 ef1d021 d1fa8b0 13e5718 bf4f857 efdc12d bf4f857 13e5718 bf4f857 13e5718 d1fa8b0 | 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 | # libraries imports
from fastapi import FastAPI, HTTPException, APIRouter
from fastapi.middleware.cors import CORSMiddleware
import asyncio
from fastapi import HTTPException
from asyncio import TimeoutError
# files imports
from apis.reddit_apis import router as reddit_router
from apis.user import router as user_router
from databases.firebase_db import initialize_firebase
from reddit.reddit_gemini import getKeywords
from utils import time_execution
# os.environ["HF_HOME"] = "./cache"
app = FastAPI(
debug=True,
title="Analysis Server",
consumes=["application/x-www-form-urlencoded", "multipart/form-data"],
docs_url='/swagger'
)
router = APIRouter()
# Assuming you have defined the necessary imports, e.g., config, getKeywords, api_key, api_key2
# CORS configuration
origins = [
"*",
# Add more origins as needed
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins, # Allows specified origins
allow_credentials=True,
allow_methods=["*"], # Allows all HTTP methods
allow_headers=["*"], # Allows all headers
)
initialize_firebase()
@router.get("/")
@time_execution
def read_root():
return {"message": "Hello, World!"}
# Timeout handler: check if getKeywords takes too long
async def fetch_keywords_with_timeout(user_query: str, timeout: int = 60, retry: bool = True):
try:
# Simulate the getKeywords function with a timeout using asyncio.wait_for
keywords = await asyncio.wait_for(asyncio.to_thread(getKeywords, user_query), timeout=timeout)
return keywords
except TimeoutError:
print("Timeout exceeded, switching to api_key2")
if retry:
# Timeout exceeded, switch to api_key2 and retry fetching keywords
read_root() # Switch API key
# Retry fetching keywords
return await fetch_keywords_with_timeout(user_query, timeout, retry=False) # Set retry to False to prevent infinite loop
else:
# If we already tried once, handle as a failure or return a fallback response
raise HTTPException(status_code=504, detail="Request timed out even after retrying")
@router.get("/keywords")
@time_execution
async def fetch_keywords(user_query: str):
if not user_query:
raise HTTPException(status_code=400, detail="User query must not be empty")
# Fetch keywords with a 10-second timeout and retry mechanism
keywords = await fetch_keywords_with_timeout(user_query=user_query)
return keywords
# if __name__ == "__main__":
# import uvicorn
# uvicorn.run("main:app", host="127.0.0.23", workers=1,reload=True,port=786)
app.include_router(router)
app.include_router(reddit_router)
app.include_router(user_router)
|