Method _send_with_auth_retry

src/openai/_client.py:440–468

(
        self,
        request: httpx.Request,
        *,
        stream: bool,
        retried: bool = False,
        **kwargs: Unpack[HttpxSendArgs],
    )

        return Querystring(array_format="brackets")

    def _send_with_auth_retry(
        self,
        request: httpx.Request,
        *,
        stream: bool,
        retried: bool = False,
        **kwargs: Unpack[HttpxSendArgs],
    ) -> httpx.Response:
        used_workload_identity_auth = False

        if self._workload_identity_auth is not None:
            authorization = request.headers.get("Authorization")
            if authorization == f"Bearer {WORKLOAD_IDENTITY_API_KEY_PLACEHOLDER}":
                request.headers["Authorization"] = f"Bearer {self._workload_identity_auth.get_token()}"
                used_workload_identity_auth = True

        response = super()._send_request(request, stream=stream, **kwargs)
        if (
            response.status_code == 401
            and self._workload_identity_auth is not None
            and used_workload_identity_auth
            and not retried
        ):
            response.close()
            self._workload_identity_auth.invalidate_token()
            request.headers["Authorization"] = f"Bearer {self._workload_identity_auth.get_token()}"
            return self._send_with_auth_retry(request, stream=stream, retried=True, **kwargs)

        return response

    @override
    def _send_request(