Spaces:
Sleeping
Sleeping
| import httpx | |
| from core.conf import settings | |
| from typing import Optional, Dict, Any | |
| from core.token_manager import get_access_token | |
| API_BASE_URL = settings.API_BASE_URL | |
| class API(): | |
| async def get(endpoint: str, params: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, api_base: str = None): | |
| if api_base: | |
| url = f"{api_base}{endpoint}" | |
| else: | |
| url = f"{API_BASE_URL}{endpoint}" | |
| access_token = await get_access_token() | |
| headers = headers or {} | |
| headers.setdefault("Content-Type", "application/json") | |
| headers["Authorization"] = f"Bearer {access_token}" | |
| async with httpx.AsyncClient() as client: | |
| try: | |
| response = await client.get(url, headers=headers, params=params) | |
| response.raise_for_status() | |
| return response.json() | |
| # except httpx.HTTPStatusError as http_err: | |
| # return {"error": f"HTTP {http_err.response.status_code}: {http_err.response.text}"} | |
| except httpx.HTTPStatusError as http_err: | |
| ## RETRY FOR 403 ERROR | |
| if http_err.response.status_code == 403: | |
| headers.pop("Authorization", None) | |
| headers["x-access-token"] = access_token | |
| headers["token_type"] = "anonymous" | |
| try: | |
| response = await client.get(url, headers=headers, params=params) | |
| response.raise_for_status() | |
| return response.json() | |
| except httpx.HTTPStatusError as second_err: | |
| return {"error": f"[403 RETRY] HTTP {second_err.response.status_code}: {second_err.response.text}"} | |
| except Exception as second_err: | |
| return {"error": f"[403 RETRY] Request failed: {str(second_err)}"} | |
| else: | |
| return {"error": f"HTTP {http_err.response.status_code}: {http_err.response.text}"} | |
| except Exception as err: | |
| return {"error": f"Request failed: {str(err)}"} | |
| async def post(endpoint: str , payload: Dict[str, Any] = None,headers: Optional[Dict[str, str]] = None, api_base: str = None): | |
| if api_base: | |
| url = f"{api_base}{endpoint}" | |
| else: | |
| url = f"{API_BASE_URL}{endpoint}" | |
| access_token = await get_access_token() | |
| headers = headers or {} | |
| headers.setdefault("Content-Type", "application/json") | |
| headers["Authorization"] = f"Bearer {access_token}" | |
| async with httpx.AsyncClient() as client: | |
| try: | |
| response = await client.post(url, json=payload, headers=headers) | |
| response.raise_for_status() | |
| return response.json() | |
| except httpx.HTTPStatusError as http_err: | |
| return {"error": f"HTTP {http_err.response.status_code}: {http_err.response.text}"} | |
| except Exception as err: | |
| return {"error": f"Request failed: {str(err)}"} | |
| async def put_api(endpoint: str, payload: Dict[str, Any], headers: Optional[Dict[str, str]] = None): | |
| url = f"{API_BASE_URL}{endpoint}" | |
| access_token = await get_access_token() | |
| headers = headers or {} | |
| headers.setdefault("Content-Type", "application/json") | |
| headers["Authorization"] = f"Bearer {access_token}" | |
| async with httpx.AsyncClient() as client: | |
| try: | |
| response = await client.put(url, json=payload, headers=headers) | |
| response.raise_for_status() | |
| return response.json() | |
| except httpx.HTTPStatusError as http_err: | |
| return {"error": f"HTTP {http_err.response.status_code}: {http_err.response.text}"} | |
| except Exception as err: | |
| return {"error": f"Request failed: {str(err)}"} | |
| async def delete_api(endpoint: str, params: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None): | |
| url = f"{API_BASE_URL}{endpoint}" | |
| access_token = await get_access_token() | |
| headers = headers or {} | |
| headers.setdefault("Content-Type", "application/json") | |
| headers["Authorization"] = f"Bearer {access_token}" | |
| async with httpx.AsyncClient() as client: | |
| try: | |
| response = await client.delete(url, headers=headers, params=params) | |
| response.raise_for_status() | |
| return response.json() | |
| except httpx.HTTPStatusError as http_err: | |
| return {"error": f"HTTP {http_err.response.status_code}: {http_err.response.text}"} | |
| except Exception as err: | |
| return {"error": f"Request failed: {str(err)}"} | |
| api: API = API() |