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.
    """
    if json_body is not None and files is not None:
        raise ValueError("json_body and files are mutually exclusive; provide one or neither.")

    pol = policy or RetryPolicy()
    exp: Tuple[int, ...] = (
        (expected_status,) if isinstance(expected_status, int) else tuple(expected_status)
    )

    # Merge headers + optional request id
    merged_headers = dict(headers or {})
    if request_id and "X-Request-ID" not in merged_headers:
        merged_headers["X-Request-ID"] = request_id

    attempt = 0
    while True:
        attempt += 1
        try:
            resp = client.request(
                method.upper(),
                url,
                params=params,
                json=json_body,
                files=files,
                headers=merged_headers or None,
                timeout=timeout,
            )
            if resp.status_code not in exp:
                if _should_retry(policy=pol, attempt=attempt, method=method, exc=None, resp=resp):
                    _sleep_with_backoff(pol, attempt)
                    continue
                raise T3HTTPError(_format_http_error_message(resp), response=resp)

            if resp.status_code == 204:
                return None
            if not resp.content:
                return None
            try:
                return resp.json()
            except json.JSONDecodeError as e:
                raise T3HTTPError("Failed to decode JSON response.", response=resp) from e
        except httpx.HTTPError as e:
            if _should_retry(policy=pol, attempt=attempt, method=method, exc=e, resp=None):
                _sleep_with_backoff(pol, attempt)
                continue
            raise T3HTTPError(str(e)) from e

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.
    """
    if json_body is not None and files is not None:
        raise ValueError("json_body and files are mutually exclusive; provide one or neither.")

    pol = policy or RetryPolicy()
    exp: Tuple[int, ...] = (
        (expected_status,) if isinstance(expected_status, int) else tuple(expected_status)
    )

    # Merge headers + optional request id
    merged_headers = dict(headers or {})
    if request_id and "X-Request-ID" not in merged_headers:
        merged_headers["X-Request-ID"] = request_id

    attempt = 0
    while True:
        attempt += 1
        try:
            resp = await aclient.request(
                method.upper(),
                url,
                params=params,
                json=json_body,
                files=files,
                headers=merged_headers or None,
                timeout=timeout,
            )
            if resp.status_code not in exp:
                if _should_retry(policy=pol, attempt=attempt, method=method, exc=None, resp=resp):
                    await _async_sleep_with_backoff(pol, attempt)
                    continue
                raise T3HTTPError(_format_http_error_message(resp), response=resp)

            if resp.status_code == 204:
                return None
            if not resp.content:
                return None
            try:
                return resp.json()
            except json.JSONDecodeError as e:
                raise T3HTTPError("Failed to decode JSON response.", response=resp) from e
        except httpx.HTTPError as e:
            if _should_retry(policy=pol, attempt=attempt, method=method, exc=e, resp=None):
                await _async_sleep_with_backoff(pol, attempt)
                continue
            raise T3HTTPError(str(e)) from e

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"]

HTTP¶

utils ¶

Scope (by design)¶

Highlights¶

Examples¶

HTTPConfig dataclass ¶

ssl_context property ¶

RetryPolicy dataclass ¶

LoggingHooks dataclass ¶

from_env classmethod ¶

as_hooks ¶

T3HTTPError ¶

status_code property ¶

build_client ¶

build_async_client ¶

request_json ¶

arequest_json async ¶

set_bearer_token ¶

clear_bearer_token ¶

HTTPConfig `dataclass` ¶

ssl_context `property` ¶

RetryPolicy `dataclass` ¶

LoggingHooks `dataclass` ¶

from_env `classmethod` ¶

status_code `property` ¶

arequest_json `async` ¶