HTTP

utils

Reusable HTTP utilities for t3api-utils (sync + async).

Scope (by design)

Configures and performs network activity (clients, retries, JSON handling, headers, SSL, proxies)

Highlights

• Centralized httpx client builders (sync + async) with sane defaults (timeout, HTTP/2, SSL via certifi, base headers, optional proxies).
• Lightweight retry policy with exponential backoff + jitter.
• Standard JSON request helpers with consistent error text.
• Simple helpers to attach/remove Bearer tokens without performing auth.
• Optional request/response logging hooks.

Examples

Sync client with bearer token: from t3api_utils.http import build_client, set_bearer_token, request_json

client = build_client()
set_bearer_token(client=client, token="<token>")
data = request_json(client=client, method="GET", url="/v2/auth/whoami")

Async with logging hooks

from t3api_utils.http import build_async_client, arequest_json, LoggingHooks

hooks = LoggingHooks(enabled=True) async with build_async_client(hooks=hooks) as aclient: data = await arequest_json(aclient=aclient, method="GET", url="/healthz")

HTTPConfig dataclass

HTTPConfig(host: str = _get_default_host(), timeout: float = DEFAULT_TIMEOUT, verify_ssl: Union[bool, str] = certifi.where(), base_headers: Mapping[str, str] = (lambda: {'User-Agent': DEFAULT_USER_AGENT})(), proxies: Optional[Union[str, Mapping[str, str]]] = None)

Base HTTP client configuration (no routes).

Attributes:

Name	Type	Description
host	str	Base URL of the API server (e.g. "https://api.example.com"). Defaults to the value returned by the config manager.
timeout	float	Request timeout in seconds. Defaults to DEFAULT_TIMEOUT.
verify_ssl	Union[bool, str]	SSL verification setting. Pass True to use default CA bundle, False to disable verification, or a file path to a custom CA certificate. Defaults to the certifi CA bundle.
base_headers	Mapping[str, str]	Default headers attached to every request built by this configuration. Defaults to a User-Agent header.
proxies	Optional[Union[str, Mapping[str, str]]]	Optional proxy URL string or mapping of scheme to proxy URL (e.g. {"https://": "http://proxy:8080"}).

ssl_context property

ssl_context: Union[bool, SSLContext]

Get proper SSL context for httpx.

RetryPolicy dataclass

RetryPolicy(max_attempts: int = 3, backoff_factor: float = 0.5, retry_methods: Sequence[str] = ('GET', 'HEAD', 'OPTIONS', 'DELETE', 'PUT', 'PATCH', 'POST'), retry_statuses: Sequence[int] = (408, 409, 425, 429, 500, 502, 503, 504))

Retry policy for transient failures. Route-agnostic.

Note: writes (POST/PUT/PATCH/DELETE) are included by default. If your call is not idempotent, provide a custom policy at the callsite.

Attributes:

Name	Type	Description
max_attempts	int	Maximum number of attempts (including the initial request). Defaults to 3.
backoff_factor	float	Base delay in seconds for exponential backoff. Actual sleep is backoff_factor * 2^(attempt - 2) with +/- 20 % jitter. Defaults to 0.5.
retry_methods	Sequence[str]	HTTP methods eligible for automatic retry. Defaults to all standard methods.
retry_statuses	Sequence[int]	HTTP status codes that trigger a retry. Defaults to common transient-error codes (408, 409, 425, 429, 500, 502, 503, 504).

LoggingHooks dataclass

LoggingHooks(enabled: bool = False, log_headers: bool = True, log_body: bool = True, file_path: Optional[str] = None)

Optional request/response logging via httpx event hooks.

Attributes:

Name	Type	Description
enabled	bool	When True, debug-level log messages are emitted for every outgoing request and incoming response. Defaults to False.
log_headers	bool	When True and logging is enabled, request headers are included in log output. Sensitive headers are masked.
log_body	bool	When True and logging is enabled, request bodies are included in log output (JSON is pretty-printed).
file_path	Optional[str]	Optional file path for HTTP debug logs. The file is opened in write mode (truncated) on first use.

from_env classmethod

from_env() -> LoggingHooks

Create a LoggingHooks instance configured from environment variables.

Reads T3_LOG_HTTP, T3_LOG_HEADERS, T3_LOG_BODY, and T3_LOG_FILE from the environment.

Returns:

Type	Description
LoggingHooks	A configured LoggingHooks instance. If T3_LOG_HTTP is
LoggingHooks	not set or is falsy, returns a disabled instance.

Source code in t3api_utils/http/utils.py

@classmethod
def from_env(cls) -> LoggingHooks:
    """Create a ``LoggingHooks`` instance configured from environment variables.

    Reads ``T3_LOG_HTTP``, ``T3_LOG_HEADERS``, ``T3_LOG_BODY``, and
    ``T3_LOG_FILE`` from the environment.

    Returns:
        A configured ``LoggingHooks`` instance. If ``T3_LOG_HTTP`` is
        not set or is falsy, returns a disabled instance.
    """
    import os
    enabled = os.getenv("T3_LOG_HTTP", "").lower() in ("true", "1", "yes", "on")
    if not enabled:
        return cls(enabled=False)

    log_headers = os.getenv("T3_LOG_HEADERS", "true").lower() not in ("false", "0", "no", "off")
    log_body = os.getenv("T3_LOG_BODY", "true").lower() not in ("false", "0", "no", "off")
    file_path = os.getenv("T3_LOG_FILE", "").strip() or None

    return cls(
        enabled=True,
        log_headers=log_headers,
        log_body=log_body,
        file_path=file_path,
    )

as_hooks

as_hooks(*, async_client: bool = False) -> Optional[Dict[str, Any]]

Build an httpx event_hooks mapping for request/response logging.

Parameters:

Name	Type	Description	Default
async_client	bool	When True, returns async hook callables suitable for httpx.AsyncClient. When False (default), returns synchronous callables for httpx.Client.	False

Returns:

Type	Description
Optional[Dict[str, Any]]	A dict with "request" and "response" hook lists, or
Optional[Dict[str, Any]]	None if logging is disabled.

Source code in t3api_utils/http/utils.py

def as_hooks(self, *, async_client: bool = False) -> Optional[Dict[str, Any]]:
    """Build an httpx ``event_hooks`` mapping for request/response logging.

    Args:
        async_client: When ``True``, returns async hook callables
            suitable for ``httpx.AsyncClient``. When ``False``
            (default), returns synchronous callables for
            ``httpx.Client``.

    Returns:
        A dict with ``"request"`` and ``"response"`` hook lists, or
        ``None`` if logging is disabled.
    """
    if not self.enabled:
        return None

    debug_logger = self._get_logger()
    do_headers = self.log_headers
    do_body = self.log_body

    def _build_request_message(request: httpx.Request) -> str:
        parts = [f">>> HTTP {request.method} {request.url}"]
        if do_headers:
            parts.append(f"  Headers:\n{_format_headers(request.headers)}")
        if do_body and request.content:
            parts.append(f"  Body:\n{_format_body(request.content)}")
        return "\n".join(parts)

    def _build_response_message(response: httpx.Response) -> str:
        req = response.request
        return f"<<< HTTP {req.method} {req.url} -> {response.status_code}"

    async def _alog_request(request: httpx.Request) -> None:
        debug_logger.debug(_build_request_message(request))

    async def _alog_response(response: httpx.Response) -> None:
        debug_logger.debug(_build_response_message(response))

    def _log_request(request: httpx.Request) -> None:
        debug_logger.debug(_build_request_message(request))

    def _log_response(response: httpx.Response) -> None:
        debug_logger.debug(_build_response_message(response))

    if async_client:
        return {
            "request": [_alog_request],
            "response": [_alog_response],
        }
    else:
        return {
            "request": [_log_request],
            "response": [_log_response],
        }

T3HTTPError

T3HTTPError(message: str, *, response: Optional[Response] = None)

Bases: HTTPError

Raised when a request fails permanently or response parsing fails.

Initialize a T3HTTPError.

Parameters:

Name	Type	Description	Default
message	str	Human-readable description of the failure.	required
response	Optional[Response]	The httpx.Response that caused the error, if available. Retained for callers that need to inspect status codes or response bodies.	None

Source code in t3api_utils/http/utils.py

def __init__(
    self, message: str, *, response: Optional[httpx.Response] = None
) -> None:
    """Initialize a T3HTTPError.

    Args:
        message: Human-readable description of the failure.
        response: The ``httpx.Response`` that caused the error, if
            available. Retained for callers that need to inspect
            status codes or response bodies.
    """
    super().__init__(message)
    self.response = response

status_code property

status_code: Optional[int]

HTTP status code from the stored response, or None if unavailable.

build_client

build_client(*, config: Optional[HTTPConfig] = None, headers: Optional[Mapping[str, str]] = None, hooks: Optional[LoggingHooks] = None) -> httpx.Client

Construct a configured httpx.Client with sane defaults.

Parameters:

Name	Type	Description	Default
config	Optional[HTTPConfig]	HTTP configuration (host, timeout, SSL, proxies). Defaults to HTTPConfig() when None.	None
headers	Optional[Mapping[str, str]]	Extra headers merged on top of config.base_headers.	None
hooks	Optional[LoggingHooks]	Optional logging hooks attached as httpx event hooks.	None

Returns:

Type	Description
Client	A ready-to-use httpx.Client instance.

Source code in t3api_utils/http/utils.py

def build_client(
    *,
    config: Optional[HTTPConfig] = None,
    headers: Optional[Mapping[str, str]] = None,
    hooks: Optional[LoggingHooks] = None,
) -> httpx.Client:
    """Construct a configured ``httpx.Client`` with sane defaults.

    Args:
        config: HTTP configuration (host, timeout, SSL, proxies).
            Defaults to ``HTTPConfig()`` when ``None``.
        headers: Extra headers merged on top of ``config.base_headers``.
        hooks: Optional logging hooks attached as httpx event hooks.

    Returns:
        A ready-to-use ``httpx.Client`` instance.
    """
    cfg = config or HTTPConfig()
    merged_headers = _merge_headers(cfg.base_headers, headers)

    return httpx.Client(
        base_url=cfg.host.rstrip("/"),
        timeout=cfg.timeout,
        verify=cfg.ssl_context,
        headers=merged_headers,
        proxy=cfg.proxies,  # type: ignore[arg-type]
        http2=False,
        event_hooks=(hooks.as_hooks(async_client=False) if hooks else None),
    )

build_async_client

build_async_client(*, config: Optional[HTTPConfig] = None, headers: Optional[Mapping[str, str]] = None, hooks: Optional[LoggingHooks] = None) -> httpx.AsyncClient

Construct a configured httpx.AsyncClient with sane defaults.

Parameters:

Name	Type	Description	Default
config	Optional[HTTPConfig]	HTTP configuration (host, timeout, SSL, proxies). Defaults to HTTPConfig() when None.	None
headers	Optional[Mapping[str, str]]	Extra headers merged on top of config.base_headers.	None
hooks	Optional[LoggingHooks]	Optional logging hooks attached as httpx event hooks.	None

Returns:

Type	Description
AsyncClient	A ready-to-use httpx.AsyncClient instance. Should be used as
AsyncClient	an async context manager or closed explicitly when finished.

Source code in t3api_utils/http/utils.py

def build_async_client(
    *,
    config: Optional[HTTPConfig] = None,
    headers: Optional[Mapping[str, str]] = None,
    hooks: Optional[LoggingHooks] = None,
) -> httpx.AsyncClient:
    """Construct a configured ``httpx.AsyncClient`` with sane defaults.

    Args:
        config: HTTP configuration (host, timeout, SSL, proxies).
            Defaults to ``HTTPConfig()`` when ``None``.
        headers: Extra headers merged on top of ``config.base_headers``.
        hooks: Optional logging hooks attached as httpx event hooks.

    Returns:
        A ready-to-use ``httpx.AsyncClient`` instance. Should be used as
        an async context manager or closed explicitly when finished.
    """
    cfg = config or HTTPConfig()
    merged_headers = _merge_headers(cfg.base_headers, headers)

    return httpx.AsyncClient(
        base_url=cfg.host.rstrip("/"),
        timeout=cfg.timeout,
        verify=cfg.ssl_context,
        headers=merged_headers,
        proxy=cfg.proxies,  # type: ignore[arg-type]
        http2=False,
        event_hooks=(hooks.as_hooks(async_client=True) if hooks else None),
    )

request_json

request_json(*, client: Client, method: str, url: str, params: Optional[Mapping[str, Any]] = None, json_body: Optional[Any] = None, files: Optional[RequestFiles] = None, headers: Optional[Mapping[str, str]] = None, policy: Optional[RetryPolicy] = None, expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204), timeout: Optional[Union[float, Timeout]] = None, request_id: Optional[str] = None) -> Any

Issue a synchronous JSON request with automatic retries.

Sends the request via the provided httpx.Client, retrying on transient failures according to the supplied (or default) retry policy. The response body is parsed as JSON and returned.

Parameters:

Name	Type	Description	Default
client	Client	An httpx.Client (typically from :func:build_client).	required
method	str	HTTP method (e.g. "GET", "POST").	required
url	str	Request URL or path (resolved against the client's base URL).	required
params	Optional[Mapping[str, Any]]	Optional query-string parameters.	None
json_body	Optional[Any]	Optional JSON-serializable request body. Mutually exclusive with files.	None
files	Optional[RequestFiles]	Optional multipart file upload data. Mutually exclusive with json_body. Accepts the same formats as httpx's files parameter (e.g. {"field": ("name.png", data, "image/png")}).	None
headers	Optional[Mapping[str, str]]	Optional per-request headers merged on top of the client's default headers.	None
policy	Optional[RetryPolicy]	Retry policy. Defaults to RetryPolicy() when None.	None
expected_status	Union[int, Iterable[int]]	Status code(s) considered successful. Defaults to (200, 201, 202, 204).	(200, 201, 202, 204)
timeout	Optional[Union[float, Timeout]]	Per-request timeout override. None uses the client default.	None
request_id	Optional[str]	Optional value set as the X-Request-ID header (unless already present in headers).	None

Returns:

Type	Description
Any	Parsed JSON response body, or None for 204 / empty responses.

Raises:

Type	Description
ValueError	If both json_body and files are provided.
T3HTTPError	If the response status is not in expected_status after all retries are exhausted, or if the response body cannot be decoded as JSON.

Source code in t3api_utils/http/utils.py

def request_json(
    *,
    client: httpx.Client,
    method: str,
    url: str,
    params: Optional[Mapping[str, Any]] = None,
    json_body: Optional[Any] = None,
    files: Optional[RequestFiles] = None,
    headers: Optional[Mapping[str, str]] = None,
    policy: Optional[RetryPolicy] = None,
    expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204),
    timeout: Optional[Union[float, httpx.Timeout]] = None,
    request_id: Optional[str] = None,
) -> Any:
    """Issue a synchronous JSON request with automatic retries.

    Sends the request via the provided ``httpx.Client``, retrying on
    transient failures according to the supplied (or default) retry
    policy. The response body is parsed as JSON and returned.

    Args:
        client: An ``httpx.Client`` (typically from :func:`build_client`).
        method: HTTP method (e.g. ``"GET"``, ``"POST"``).
        url: Request URL or path (resolved against the client's base URL).
        params: Optional query-string parameters.
        json_body: Optional JSON-serializable request body. Mutually
            exclusive with *files*.
        files: Optional multipart file upload data. Mutually exclusive
            with *json_body*. Accepts the same formats as ``httpx``'s
            ``files`` parameter (e.g.
            ``{"field": ("name.png", data, "image/png")}``).
        headers: Optional per-request headers merged on top of the
            client's default headers.
        policy: Retry policy. Defaults to ``RetryPolicy()`` when ``None``.
        expected_status: Status code(s) considered successful. Defaults
            to ``(200, 201, 202, 204)``.
        timeout: Per-request timeout override. ``None`` uses the client
            default.
        request_id: Optional value set as the ``X-Request-ID`` header
            (unless already present in *headers*).

    Returns:
        Parsed JSON response body, or ``None`` for 204 / empty responses.

    Raises:
        ValueError: If both *json_body* and *files* are provided.
        T3HTTPError: If the response status is not in *expected_status*
            after all retries are exhausted, or if the response body
            cannot be decoded as JSON.
    """
    resp = _request_core(
        client=client,
        method=method,
        url=url,
        params=params,
        json_body=json_body,
        files=files,
        headers=headers,
        policy=policy,
        expected_status=expected_status,
        timeout=timeout,
        request_id=request_id,
    )
    return _parse_json_response(resp)

request_bytes

request_bytes(*, client: Client, method: str, url: str, params: Optional[Mapping[str, Any]] = None, json_body: Optional[Any] = None, files: Optional[RequestFiles] = None, headers: Optional[Mapping[str, str]] = None, policy: Optional[RetryPolicy] = None, expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204), timeout: Optional[Union[float, Timeout]] = None, request_id: Optional[str] = None) -> bytes

Issue a synchronous request and return the response body as bytes.

Same retry and error handling as :func:request_json, but returns raw bytes instead of parsing JSON. Useful for downloading PDFs, images, and other binary content.

Returns:

Type	Description
bytes	Response body as bytes. Empty b"" for 204 / empty responses.

Source code in t3api_utils/http/utils.py

def request_bytes(
    *,
    client: httpx.Client,
    method: str,
    url: str,
    params: Optional[Mapping[str, Any]] = None,
    json_body: Optional[Any] = None,
    files: Optional[RequestFiles] = None,
    headers: Optional[Mapping[str, str]] = None,
    policy: Optional[RetryPolicy] = None,
    expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204),
    timeout: Optional[Union[float, httpx.Timeout]] = None,
    request_id: Optional[str] = None,
) -> bytes:
    """Issue a synchronous request and return the response body as bytes.

    Same retry and error handling as :func:`request_json`, but returns
    raw bytes instead of parsing JSON. Useful for downloading PDFs,
    images, and other binary content.

    Returns:
        Response body as ``bytes``. Empty ``b""`` for 204 / empty responses.
    """
    resp = _request_core(
        client=client,
        method=method,
        url=url,
        params=params,
        json_body=json_body,
        files=files,
        headers=headers,
        policy=policy,
        expected_status=expected_status,
        timeout=timeout,
        request_id=request_id,
    )
    return resp.content

request_text

request_text(*, client: Client, method: str, url: str, params: Optional[Mapping[str, Any]] = None, json_body: Optional[Any] = None, files: Optional[RequestFiles] = None, headers: Optional[Mapping[str, str]] = None, policy: Optional[RetryPolicy] = None, expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204), timeout: Optional[Union[float, Timeout]] = None, request_id: Optional[str] = None) -> str

Issue a synchronous request and return the response body as text.

Same retry and error handling as :func:request_json, but returns decoded text instead of parsing JSON. Useful for CSV, HTML, and other text content.

Returns:

Type	Description
str	Response body as str. Empty "" for 204 / empty responses.

Source code in t3api_utils/http/utils.py

def request_text(
    *,
    client: httpx.Client,
    method: str,
    url: str,
    params: Optional[Mapping[str, Any]] = None,
    json_body: Optional[Any] = None,
    files: Optional[RequestFiles] = None,
    headers: Optional[Mapping[str, str]] = None,
    policy: Optional[RetryPolicy] = None,
    expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204),
    timeout: Optional[Union[float, httpx.Timeout]] = None,
    request_id: Optional[str] = None,
) -> str:
    """Issue a synchronous request and return the response body as text.

    Same retry and error handling as :func:`request_json`, but returns
    decoded text instead of parsing JSON. Useful for CSV, HTML, and
    other text content.

    Returns:
        Response body as ``str``. Empty ``""`` for 204 / empty responses.
    """
    resp = _request_core(
        client=client,
        method=method,
        url=url,
        params=params,
        json_body=json_body,
        files=files,
        headers=headers,
        policy=policy,
        expected_status=expected_status,
        timeout=timeout,
        request_id=request_id,
    )
    return resp.text

request_raw

request_raw(*, client: Client, method: str, url: str, params: Optional[Mapping[str, Any]] = None, json_body: Optional[Any] = None, files: Optional[RequestFiles] = None, headers: Optional[Mapping[str, str]] = None, policy: Optional[RetryPolicy] = None, expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204), timeout: Optional[Union[float, Timeout]] = None, request_id: Optional[str] = None) -> httpx.Response

Issue a synchronous request and return the raw httpx.Response.

Same retry and error handling as :func:request_json, but returns the full response object for callers that need access to headers, content-type, or streaming.

Returns:

Type	Description
Response	The validated httpx.Response.

Source code in t3api_utils/http/utils.py

def request_raw(
    *,
    client: httpx.Client,
    method: str,
    url: str,
    params: Optional[Mapping[str, Any]] = None,
    json_body: Optional[Any] = None,
    files: Optional[RequestFiles] = None,
    headers: Optional[Mapping[str, str]] = None,
    policy: Optional[RetryPolicy] = None,
    expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204),
    timeout: Optional[Union[float, httpx.Timeout]] = None,
    request_id: Optional[str] = None,
) -> httpx.Response:
    """Issue a synchronous request and return the raw ``httpx.Response``.

    Same retry and error handling as :func:`request_json`, but returns
    the full response object for callers that need access to headers,
    content-type, or streaming.

    Returns:
        The validated ``httpx.Response``.
    """
    return _request_core(
        client=client,
        method=method,
        url=url,
        params=params,
        json_body=json_body,
        files=files,
        headers=headers,
        policy=policy,
        expected_status=expected_status,
        timeout=timeout,
        request_id=request_id,
    )

arequest_json async

arequest_json(*, aclient: AsyncClient, method: str, url: str, params: Optional[Mapping[str, Any]] = None, json_body: Optional[Any] = None, files: Optional[RequestFiles] = None, headers: Optional[Mapping[str, str]] = None, policy: Optional[RetryPolicy] = None, expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204), timeout: Optional[Union[float, Timeout]] = None, request_id: Optional[str] = None) -> Any

Issue an asynchronous JSON request with automatic retries.

Async equivalent of :func:request_json. Sends the request via the provided httpx.AsyncClient, retrying on transient failures according to the supplied (or default) retry policy.

Parameters:

Name	Type	Description	Default
aclient	AsyncClient	An httpx.AsyncClient (typically from :func:build_async_client).	required
method	str	HTTP method (e.g. "GET", "POST").	required
url	str	Request URL or path (resolved against the client's base URL).	required
params	Optional[Mapping[str, Any]]	Optional query-string parameters.	None
json_body	Optional[Any]	Optional JSON-serializable request body. Mutually exclusive with files.	None
files	Optional[RequestFiles]	Optional multipart file upload data. Mutually exclusive with json_body. Accepts the same formats as httpx's files parameter.	None
headers	Optional[Mapping[str, str]]	Optional per-request headers merged on top of the client's default headers.	None
policy	Optional[RetryPolicy]	Retry policy. Defaults to RetryPolicy() when None.	None
expected_status	Union[int, Iterable[int]]	Status code(s) considered successful. Defaults to (200, 201, 202, 204).	(200, 201, 202, 204)
timeout	Optional[Union[float, Timeout]]	Per-request timeout override. None uses the client default.	None
request_id	Optional[str]	Optional value set as the X-Request-ID header (unless already present in headers).	None

Returns:

Type	Description
Any	Parsed JSON response body, or None for 204 / empty responses.

Raises:

Type	Description
ValueError	If both json_body and files are provided.
T3HTTPError	If the response status is not in expected_status after all retries are exhausted, or if the response body cannot be decoded as JSON.

Source code in t3api_utils/http/utils.py

async def arequest_json(
    *,
    aclient: httpx.AsyncClient,
    method: str,
    url: str,
    params: Optional[Mapping[str, Any]] = None,
    json_body: Optional[Any] = None,
    files: Optional[RequestFiles] = None,
    headers: Optional[Mapping[str, str]] = None,
    policy: Optional[RetryPolicy] = None,
    expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204),
    timeout: Optional[Union[float, httpx.Timeout]] = None,
    request_id: Optional[str] = None,
) -> Any:
    """Issue an asynchronous JSON request with automatic retries.

    Async equivalent of :func:`request_json`. Sends the request via the
    provided ``httpx.AsyncClient``, retrying on transient failures
    according to the supplied (or default) retry policy.

    Args:
        aclient: An ``httpx.AsyncClient`` (typically from
            :func:`build_async_client`).
        method: HTTP method (e.g. ``"GET"``, ``"POST"``).
        url: Request URL or path (resolved against the client's base URL).
        params: Optional query-string parameters.
        json_body: Optional JSON-serializable request body. Mutually
            exclusive with *files*.
        files: Optional multipart file upload data. Mutually exclusive
            with *json_body*. Accepts the same formats as ``httpx``'s
            ``files`` parameter.
        headers: Optional per-request headers merged on top of the
            client's default headers.
        policy: Retry policy. Defaults to ``RetryPolicy()`` when ``None``.
        expected_status: Status code(s) considered successful. Defaults
            to ``(200, 201, 202, 204)``.
        timeout: Per-request timeout override. ``None`` uses the client
            default.
        request_id: Optional value set as the ``X-Request-ID`` header
            (unless already present in *headers*).

    Returns:
        Parsed JSON response body, or ``None`` for 204 / empty responses.

    Raises:
        ValueError: If both *json_body* and *files* are provided.
        T3HTTPError: If the response status is not in *expected_status*
            after all retries are exhausted, or if the response body
            cannot be decoded as JSON.
    """
    resp = await _arequest_core(
        aclient=aclient,
        method=method,
        url=url,
        params=params,
        json_body=json_body,
        files=files,
        headers=headers,
        policy=policy,
        expected_status=expected_status,
        timeout=timeout,
        request_id=request_id,
    )
    return _parse_json_response(resp)

arequest_bytes async

arequest_bytes(*, aclient: AsyncClient, method: str, url: str, params: Optional[Mapping[str, Any]] = None, json_body: Optional[Any] = None, files: Optional[RequestFiles] = None, headers: Optional[Mapping[str, str]] = None, policy: Optional[RetryPolicy] = None, expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204), timeout: Optional[Union[float, Timeout]] = None, request_id: Optional[str] = None) -> bytes

Async equivalent of :func:request_bytes.

Source code in t3api_utils/http/utils.py

async def arequest_bytes(
    *,
    aclient: httpx.AsyncClient,
    method: str,
    url: str,
    params: Optional[Mapping[str, Any]] = None,
    json_body: Optional[Any] = None,
    files: Optional[RequestFiles] = None,
    headers: Optional[Mapping[str, str]] = None,
    policy: Optional[RetryPolicy] = None,
    expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204),
    timeout: Optional[Union[float, httpx.Timeout]] = None,
    request_id: Optional[str] = None,
) -> bytes:
    """Async equivalent of :func:`request_bytes`."""
    resp = await _arequest_core(
        aclient=aclient,
        method=method,
        url=url,
        params=params,
        json_body=json_body,
        files=files,
        headers=headers,
        policy=policy,
        expected_status=expected_status,
        timeout=timeout,
        request_id=request_id,
    )
    return resp.content

arequest_text async

arequest_text(*, aclient: AsyncClient, method: str, url: str, params: Optional[Mapping[str, Any]] = None, json_body: Optional[Any] = None, files: Optional[RequestFiles] = None, headers: Optional[Mapping[str, str]] = None, policy: Optional[RetryPolicy] = None, expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204), timeout: Optional[Union[float, Timeout]] = None, request_id: Optional[str] = None) -> str

Async equivalent of :func:request_text.

Source code in t3api_utils/http/utils.py

async def arequest_text(
    *,
    aclient: httpx.AsyncClient,
    method: str,
    url: str,
    params: Optional[Mapping[str, Any]] = None,
    json_body: Optional[Any] = None,
    files: Optional[RequestFiles] = None,
    headers: Optional[Mapping[str, str]] = None,
    policy: Optional[RetryPolicy] = None,
    expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204),
    timeout: Optional[Union[float, httpx.Timeout]] = None,
    request_id: Optional[str] = None,
) -> str:
    """Async equivalent of :func:`request_text`."""
    resp = await _arequest_core(
        aclient=aclient,
        method=method,
        url=url,
        params=params,
        json_body=json_body,
        files=files,
        headers=headers,
        policy=policy,
        expected_status=expected_status,
        timeout=timeout,
        request_id=request_id,
    )
    return resp.text

arequest_raw async

arequest_raw(*, aclient: AsyncClient, method: str, url: str, params: Optional[Mapping[str, Any]] = None, json_body: Optional[Any] = None, files: Optional[RequestFiles] = None, headers: Optional[Mapping[str, str]] = None, policy: Optional[RetryPolicy] = None, expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204), timeout: Optional[Union[float, Timeout]] = None, request_id: Optional[str] = None) -> httpx.Response

Async equivalent of :func:request_raw.

Source code in t3api_utils/http/utils.py

async def arequest_raw(
    *,
    aclient: httpx.AsyncClient,
    method: str,
    url: str,
    params: Optional[Mapping[str, Any]] = None,
    json_body: Optional[Any] = None,
    files: Optional[RequestFiles] = None,
    headers: Optional[Mapping[str, str]] = None,
    policy: Optional[RetryPolicy] = None,
    expected_status: Union[int, Iterable[int]] = (200, 201, 202, 204),
    timeout: Optional[Union[float, httpx.Timeout]] = None,
    request_id: Optional[str] = None,
) -> httpx.Response:
    """Async equivalent of :func:`request_raw`."""
    return await _arequest_core(
        aclient=aclient,
        method=method,
        url=url,
        params=params,
        json_body=json_body,
        files=files,
        headers=headers,
        policy=policy,
        expected_status=expected_status,
        timeout=timeout,
        request_id=request_id,
    )

set_bearer_token

set_bearer_token(*, client: Union[Client, AsyncClient], token: str) -> None

Attach or replace the Authorization: Bearer header on a client.

Parameters:

Name	Type	Description	Default
client	Union[Client, AsyncClient]	Sync or async httpx client whose headers will be modified in place.	required
token	str	Raw bearer token string (without the Bearer prefix).	required

Source code in t3api_utils/http/utils.py

def set_bearer_token(*, client: Union[httpx.Client, httpx.AsyncClient], token: str) -> None:
    """Attach or replace the ``Authorization: Bearer`` header on a client.

    Args:
        client: Sync or async ``httpx`` client whose headers will be
            modified in place.
        token: Raw bearer token string (without the ``Bearer `` prefix).
    """
    client.headers["Authorization"] = f"Bearer {token}"

clear_bearer_token

clear_bearer_token(*, client: Union[Client, AsyncClient]) -> None

Remove the Authorization header from a client, if present.

Parameters:

Name	Type	Description	Default
client	Union[Client, AsyncClient]	Sync or async httpx client whose headers will be modified in place. No error is raised if the header is already absent.	required

Source code in t3api_utils/http/utils.py

def clear_bearer_token(*, client: Union[httpx.Client, httpx.AsyncClient]) -> None:
    """Remove the ``Authorization`` header from a client, if present.

    Args:
        client: Sync or async ``httpx`` client whose headers will be
            modified in place. No error is raised if the header is
            already absent.
    """
    if "Authorization" in client.headers:
        del client.headers["Authorization"]