"""HTTP client wrapping requests."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, Generic, TypeVar, cast
import requests
from .auth import Authenticator
ResponseData = TypeVar("ResponseData")
class Response:
"""Thin wrapper over requests.Response with strict types."""
def __init__(self, raw: requests.Response) -> None:
self._raw = raw
@property
def status_code(self) -> int:
return self._raw.status_code # type: ignore[return-value]
@property
def reason(self) -> str:
return self._raw.reason or ""
@property
def content(self) -> bytes:
return self._raw.content # type: ignore[return-value]
@property
def headers(self) -> Mapping[str, str]:
return cast(Mapping[str, str], self._raw.headers)
@property
def text(self) -> str:
return self._raw.text
def json(self) -> Any: # type: ignore[explicit-override]
return self._raw.json() # type: ignore[reportUnknownMemberType]
@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: Response
class Client:
"""Synchronous HTTP client for the generated SDK."""
def __init__(
self,
base_url: str,
*,
http_client: requests.Session | None = None,
authenticator: Authenticator | None = None,
) -> None:
self._base_url = base_url.rstrip("/")
self._http_client = http_client or requests.Session()
self._authenticator = authenticator
def request(
self,
method: str,
path: str,
*,
params: dict[str, str] | None = None,
json: Any = None,
data: Any = None,
files: Any = None,
headers: dict[str, str] | None = None,
) -> 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())
raw = cast(
requests.Response,
self._http_client.request( # type: ignore[reportUnknownMemberType]
method,
url,
params=params,
json=json,
data=data,
files=files,
headers=req_headers,
),
)
return Response(raw)
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()