Spaces:
Sleeping
Sleeping
File size: 3,603 Bytes
aa5cda2 bf03cdf aa5cda2 |
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 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 |
"""API client for backend communication"""
import requests
import logging
from typing import Dict, List, Any
logger = logging.getLogger(__name__)
# Import config
import os
from dotenv import load_dotenv
load_dotenv()
BACKEND_URL = os.getenv("BACKEND_URL", "http://localhost:8000")
API_BASE_URL = f"{BACKEND_URL}/api/v1"
TIMEOUT_LONG = 120
class APIClient:
"""Client for backend API communication"""
def __init__(self):
backend_url = BACKEND_URL
self.base_url = API_BASE_URL
self.timeout = TIMEOUT_LONG
def health_check(self) -> Dict[str, Any]:
"""Check backend health status"""
try:
response = requests.get(
f"{self.base_url}/health",
timeout=self.timeout
)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"❌ Health check failed: {e}")
return {"status": "error", "detail": str(e)}
def chat(self, messages: List[Dict], system_prompt: str = None) -> Dict[str, Any]:
"""Send chat request to backend"""
try:
payload = {
"messages": messages,
"system_prompt": system_prompt or "You are a helpful data analysis assistant."
}
response = requests.post(
f"{self.base_url}/chat",
json=payload,
timeout=self.timeout
)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"❌ Chat request failed: {e}")
return {"error": str(e)}
def upload_file(self, file_bytes: bytes, filename: str) -> Dict[str, Any]:
"""Upload file to backend"""
try:
files = {"file": (filename, file_bytes)}
response = requests.post(
f"{self.base_url}/upload",
files=files,
timeout=self.timeout
)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"❌ File upload failed: {e}")
return {"error": str(e)}
def analyze(self, data: List[Dict], analysis_type: str, columns: List[str] = None) -> Dict[str, Any]:
"""Request data analysis"""
try:
payload = {
"data": data,
"analysis_type": analysis_type,
"columns": columns or []
}
response = requests.post(
f"{self.base_url}/analyze",
json=payload,
timeout=self.timeout
)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"❌ Analysis request failed: {e}")
return {"error": str(e)}
def get_suggestions(self, data: List[Dict], context: str = None) -> Dict[str, Any]:
"""Get AI suggestions for data"""
try:
payload = {
"data": data,
"analysis_context": context or ""
}
response = requests.post(
f"{self.base_url}/suggestions",
json=payload,
timeout=self.timeout
)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"❌ Suggestion request failed: {e}")
return {"error": str(e)}
# Global client instance
client = APIClient()
|