"""HTTP client wrapping httpx."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, Generic, TypeVar
import httpx
from .auth import Authenticator
ResponseData = TypeVar("ResponseData")
@dataclass(frozen=True)
class ApiResponse(Generic[ResponseData]):
"""Decoded response data together with its HTTP metadata."""
data: ResponseData
status_code: int
headers: Mapping[str, str]
raw: httpx.Response
class Client:
"""Synchronous HTTP client for the generated SDK."""
def __init__(
self,
base_url: str,
*,
http_client: httpx.Client | None = None,
authenticator: Authenticator | None = None,
) -> None:
self._base_url = base_url.rstrip("/")
self._http_client = http_client or httpx.Client()
self._authenticator = authenticator
def request(
self,
method: str,
path: str,
*,
params: dict[str, str] | None = None,
json: Any = None,
content: Any = None,
data: Any = None,
files: Any = None,
headers: dict[str, str] | None = None,
) -> httpx.Response:
"""Send an HTTP request and return the raw response."""
url = f"{self._base_url}{path}"
req_headers: dict[str, str] = {"Accept": "application/json"}
if json is not None:
req_headers["Content-Type"] = "application/json"
if headers:
req_headers.update(headers)
if self._authenticator is not None:
req_headers.update(self._authenticator.auth_headers())
return self._http_client.request(
method,
url,
params=params,
json=json,
content=content,
data=data,
files=files,
headers=req_headers,
)
def close(self) -> None:
"""Close the underlying HTTP client."""
self._http_client.close()
def __enter__(self) -> Client:
return self
def __exit__(self, *args: object) -> None:
self.close()