mirror of
https://github.com/Av0idD4rk/packmate.git
synced 2026-10-23 15:31:00 +00:00
24 lines
1.2 KiB
Python
24 lines
1.2 KiB
Python
import httpx
|
|
|
|
|
|
class Backend:
|
|
def __init__(self, url: str, login: str, password: str, transport=None):
|
|
self.client = httpx.AsyncClient(base_url=url.rstrip("/"), auth=(login, password), timeout=8,
|
|
limits=httpx.Limits(max_connections=4, max_keepalive_connections=4),
|
|
transport=transport, trust_env=False)
|
|
|
|
async def close(self):
|
|
await self.client.aclose()
|
|
|
|
async def request(self, method: str, path: str, **kwargs):
|
|
try:
|
|
response = await self.client.request(method, path, **kwargs)
|
|
except httpx.HTTPError:
|
|
raise ValueError("BACKEND_UNAVAILABLE: request failed; writes may have committed, check current state before retry") from None
|
|
codes = {400: "INVALID_ARGUMENT", 401: "BACKEND_AUTH_FAILED", 403: "BACKEND_AUTH_FAILED",
|
|
404: "NOT_FOUND", 409: "CONFLICT", 429: "BUSY", 503: "BUSY"}
|
|
if response.is_error:
|
|
# Never send stack traces, SQL or backend credentials into an agent's context.
|
|
raise ValueError(f"{codes.get(response.status_code, 'BACKEND_ERROR')}: HTTP {response.status_code}")
|
|
return response.json() if response.content else {"ok": True}
|