33 lines
1.2 KiB
Python
33 lines
1.2 KiB
Python
from typing import Any, Dict, Optional
|
|
import httpx
|
|
|
|
|
|
class ApiClient:
|
|
def __init__(self, base_url: Optional[str], api_key: Optional[str] = None):
|
|
self.base_url = base_url.rstrip("/") if base_url else None
|
|
self.api_key = api_key
|
|
|
|
def configured(self) -> bool:
|
|
return bool(self.base_url)
|
|
|
|
def headers(self) -> Dict[str, str]:
|
|
return {"X-Api-Key": self.api_key} if self.api_key else {}
|
|
|
|
async def get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Optional[Any]:
|
|
if not self.base_url:
|
|
return None
|
|
url = f"{self.base_url}{path}"
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.get(url, headers=self.headers(), params=params)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def post(self, path: str, payload: Optional[Dict[str, Any]] = None) -> Optional[Any]:
|
|
if not self.base_url:
|
|
return None
|
|
url = f"{self.base_url}{path}"
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.post(url, headers=self.headers(), json=payload)
|
|
response.raise_for_status()
|
|
return response.json()
|