49 lines
1.8 KiB
Python
49 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
USER_AGENT = "OpenHerbarium-MCP/0.1 (+https://example.invalid/openherbarium-mcp)"
|
|
|
|
|
|
class BotanicalHTTPClient:
|
|
"""Small async HTTP wrapper with conservative defaults for public botanical APIs."""
|
|
|
|
def __init__(self, timeout: float = 20.0, transport: httpx.AsyncBaseTransport | None = None):
|
|
self.timeout = timeout
|
|
self.transport = transport
|
|
self._client: httpx.AsyncClient | None = None
|
|
|
|
async def __aenter__(self) -> "BotanicalHTTPClient":
|
|
self._client = httpx.AsyncClient(
|
|
timeout=self.timeout,
|
|
follow_redirects=True,
|
|
transport=self.transport,
|
|
headers={"User-Agent": USER_AGENT, "Accept": "application/json,text/html;q=0.9,*/*;q=0.8"},
|
|
)
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb) -> None: # type: ignore[no-untyped-def]
|
|
if self._client:
|
|
await self._client.aclose()
|
|
|
|
@property
|
|
def client(self) -> httpx.AsyncClient:
|
|
if self._client is None:
|
|
raise RuntimeError("BotanicalHTTPClient must be used as an async context manager")
|
|
return self._client
|
|
|
|
async def get_json(self, url: str, params: Mapping[str, Any] | None = None) -> dict[str, Any]:
|
|
response = await self.client.get(url, params=params)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return data if isinstance(data, dict) else {"results": data}
|
|
|
|
async def post_json(self, url: str, payload: Mapping[str, Any], headers: Mapping[str, str] | None = None) -> dict[str, Any]:
|
|
response = await self.client.post(url, json=payload, headers=headers)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return data if isinstance(data, dict) else {"results": data}
|