| |
| 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() |
|
|
