|
|
|
|
|
import os, httpx |
|
|
from functools import lru_cache |
|
|
from typing import Any, Dict |
|
|
|
|
|
class APIError(Exception): |
|
|
pass |
|
|
|
|
|
class BaseClient: |
|
|
def __init__(self, base_url: str, api_key_env: str = None): |
|
|
self.base = base_url.rstrip("/") |
|
|
self.key = os.getenv(api_key_env) if api_key_env else None |
|
|
self._client = httpx.AsyncClient(timeout=10) |
|
|
|
|
|
async def request(self, method: str, path: str, **kwargs) -> Any: |
|
|
url = f"{self.base}/{path.lstrip('/')}" |
|
|
headers = kwargs.pop("headers", {}) |
|
|
if self.key: |
|
|
headers["Authorization"] = f"Bearer {self.key}" |
|
|
resp = await self._client.request(method, url, headers=headers, **kwargs) |
|
|
try: |
|
|
resp.raise_for_status() |
|
|
except httpx.HTTPStatusError as e: |
|
|
raise APIError(f"{resp.status_code} @ {url}") from e |
|
|
ct = resp.headers.get("Content-Type", "") |
|
|
return resp.json() if ct.startswith("application/json") else resp.text |
|
|
|
|
|
async def close(self): |
|
|
await self._client.aclose() |
|
|
|
