Spaces:
Sleeping
Sleeping
File size: 2,805 Bytes
e2b74e4 b24ccd7 e2b74e4 |
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 |
import httpx
from app.core.config import settings
from typing import Optional, Dict, Any
API_BASE_URL = settings.API_BASE_URL
async def get(endpoint: str, params: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None):
"""Gọi API GET"""
url = f"{API_BASE_URL}{endpoint}"
headers = headers or {}
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 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):
"""Gọi API POST"""
url = f"{API_BASE_URL}{endpoint}"
headers = headers or {"Content-Type": "application/json"}
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):
"""Gọi API PUT"""
url = f"{API_BASE_URL}{endpoint}"
headers = headers or {"Content-Type": "application/json"}
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):
"""Gọi API DELETE"""
url = f"{API_BASE_URL}{endpoint}"
headers = headers or {}
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)}"}
|