# This file was auto-generated by Fern from our API Definition.

import asyncio
import email.utils
import re
import socket
import time
import typing
from contextlib import asynccontextmanager, contextmanager
from random import random

import httpx
from .file import File, convert_file_dict_to_httpx_tuples
from .force_multipart import FORCE_MULTIPART
from .jsonable_encoder import jsonable_encoder
from .logging import LogConfig, Logger, create_logger
from .query_encoder import encode_query
from .remove_none_from_dict import remove_none_from_dict as remove_none_from_dict
from .request_options import RequestOptions
from httpx._types import RequestFiles

INITIAL_RETRY_DELAY_SECONDS = 1.0
MAX_RETRY_DELAY_SECONDS = 60.0
JITTER_FACTOR = 0.2  # 20% random jitter


def get_keepalive_socket_options(
    idle: int = 60,
    intvl: int = 30,
    cnt: int = 5,
) -> typing.List[typing.Tuple[int, int, int]]:
    """
    Build TCP keepalive socket options for the current platform.

    Keepalive probes keep otherwise-idle connections alive so that long,
    non-streaming requests survive idle-connection reaping by a firewall,
    load balancer, or NAT. The available socket constants are OS-dependent,
    so each option is guarded and only emitted when the platform defines it:

    - ``SO_KEEPALIVE`` is portable (Linux/macOS/Windows).
    - The idle-before-first-probe knob is ``TCP_KEEPIDLE`` on Linux and modern
      Windows, but ``TCP_KEEPALIVE`` on macOS.
    - ``TCP_KEEPINTVL`` / ``TCP_KEEPCNT`` exist on Linux/macOS/modern Windows.

    Passing these tuples to ``httpx.HTTPTransport(socket_options=...)`` /
    ``httpx.AsyncHTTPTransport(socket_options=...)`` applies them to every
    connection the transport opens.
    """
    opts: typing.List[typing.Tuple[int, int, int]] = [(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)]
    idle_const = getattr(socket, "TCP_KEEPIDLE", None) or getattr(socket, "TCP_KEEPALIVE", None)
    if idle_const:
        opts.append((socket.IPPROTO_TCP, idle_const, idle))
    if hasattr(socket, "TCP_KEEPINTVL"):
        opts.append((socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, intvl))
    if hasattr(socket, "TCP_KEEPCNT"):
        opts.append((socket.IPPROTO_TCP, socket.TCP_KEEPCNT, cnt))
    return opts


def _parse_retry_after(response_headers: httpx.Headers) -> typing.Optional[float]:
    """
    This function parses the `Retry-After` header in a HTTP response and returns the number of seconds to wait.

    Inspired by the urllib3 retry implementation.
    """
    retry_after_ms = response_headers.get("retry-after-ms")
    if retry_after_ms is not None:
        try:
            return int(retry_after_ms) / 1000 if retry_after_ms > 0 else 0
        except Exception:
            pass

    retry_after = response_headers.get("retry-after")
    if retry_after is None:
        return None

    # Attempt to parse the header as an int.
    if re.match(r"^\s*[0-9]+\s*$", retry_after):
        seconds = float(retry_after)
    # Fallback to parsing it as a date.
    else:
        retry_date_tuple = email.utils.parsedate_tz(retry_after)
        if retry_date_tuple is None:
            return None
        if retry_date_tuple[9] is None:  # Python 2
            # Assume UTC if no timezone was specified
            # On Python2.7, parsedate_tz returns None for a timezone offset
            # instead of 0 if no timezone is given, where mktime_tz treats
            # a None timezone offset as local time.
            retry_date_tuple = retry_date_tuple[:9] + (0,) + retry_date_tuple[10:]

        retry_date = email.utils.mktime_tz(retry_date_tuple)
        seconds = retry_date - time.time()

    if seconds < 0:
        seconds = 0

    return seconds


def _add_positive_jitter(delay: float) -> float:
    """Add positive jitter (0-20%) to prevent thundering herd."""
    jitter_multiplier = 1 + random() * JITTER_FACTOR
    return delay * jitter_multiplier


def _add_symmetric_jitter(delay: float) -> float:
    """Add symmetric jitter (±10%) for exponential backoff."""
    jitter_multiplier = 1 + (random() - 0.5) * JITTER_FACTOR
    return delay * jitter_multiplier


def _parse_x_ratelimit_reset(response_headers: httpx.Headers) -> typing.Optional[float]:
    """
    Parse the X-RateLimit-Reset header (Unix timestamp in seconds).
    Returns seconds to wait, or None if header is missing/invalid.
    """
    reset_time_str = response_headers.get("x-ratelimit-reset")
    if reset_time_str is None:
        return None

    try:
        reset_time = int(reset_time_str)
        delay = reset_time - time.time()
        if delay > 0:
            return delay
    except (ValueError, TypeError):
        pass

    return None


def _retry_timeout(response: httpx.Response, retries: int) -> float:
    """
    Determine the amount of time to wait before retrying a request.
    This function begins by trying to parse a retry-after header from the response, and then proceeds to use exponential backoff
    with a jitter to determine the number of seconds to wait.
    """

    # 1. Check Retry-After header first
    retry_after = _parse_retry_after(response.headers)
    if retry_after is not None and retry_after > 0:
        return min(retry_after, MAX_RETRY_DELAY_SECONDS)

    # 2. Check X-RateLimit-Reset header (with positive jitter)
    ratelimit_reset = _parse_x_ratelimit_reset(response.headers)
    if ratelimit_reset is not None:
        return _add_positive_jitter(min(ratelimit_reset, MAX_RETRY_DELAY_SECONDS))

    # 3. Fall back to exponential backoff (with symmetric jitter)
    backoff = min(INITIAL_RETRY_DELAY_SECONDS * pow(2.0, retries), MAX_RETRY_DELAY_SECONDS)
    return _add_symmetric_jitter(backoff)


def _retry_timeout_from_retries(retries: int) -> float:
    """Determine retry timeout using exponential backoff when no response is available."""
    backoff = min(INITIAL_RETRY_DELAY_SECONDS * pow(2.0, retries), MAX_RETRY_DELAY_SECONDS)
    return _add_symmetric_jitter(backoff)


def _should_retry(response: httpx.Response) -> bool:
    return response.status_code >= 500 or response.status_code in [429, 408, 409]


_SENSITIVE_HEADERS = frozenset(
    {
        "authorization",
        "www-authenticate",
        "x-api-key",
        "api-key",
        "apikey",
        "x-api-token",
        "x-auth-token",
        "auth-token",
        "cookie",
        "set-cookie",
        "proxy-authorization",
        "proxy-authenticate",
        "x-csrf-token",
        "x-xsrf-token",
        "x-session-token",
        "x-access-token",
    }
)


def _redact_headers(headers: typing.Dict[str, str]) -> typing.Dict[str, str]:
    return {k: ("[REDACTED]" if k.lower() in _SENSITIVE_HEADERS else v) for k, v in headers.items()}


def _build_url(base_url: str, path: typing.Optional[str]) -> str:
    """
    Build a full URL by joining a base URL with a path.

    This function correctly handles base URLs that contain path prefixes (e.g., tenant-based URLs)
    by using string concatenation instead of urllib.parse.urljoin(), which would incorrectly
    strip path components when the path starts with '/'.

    Example:
        >>> _build_url("https://cloud.example.com/org/tenant/api", "/users")
        'https://cloud.example.com/org/tenant/api/users'

    Args:
        base_url: The base URL, which may contain path prefixes.
        path: The path to append. Can be None or empty string.

    Returns:
        The full URL with base_url and path properly joined.
    """
    if not path:
        return base_url
    return f"{base_url.rstrip('/')}/{path.lstrip('/')}"


def _maybe_filter_none_from_multipart_data(
    data: typing.Optional[typing.Any],
    request_files: typing.Optional[RequestFiles],
    force_multipart: typing.Optional[bool],
) -> typing.Optional[typing.Any]:
    """
    Filter None values from data body for multipart/form requests.
    This prevents httpx from converting None to empty strings in multipart encoding.
    Only applies when files are present or force_multipart is True.
    """
    if data is not None and isinstance(data, typing.Mapping) and (request_files or force_multipart):
        return remove_none_from_dict(data)
    return data


def remove_omit_from_dict(
    original: typing.Dict[str, typing.Optional[typing.Any]],
    omit: typing.Optional[typing.Any],
) -> typing.Dict[str, typing.Any]:
    if omit is None:
        return original
    new: typing.Dict[str, typing.Any] = {}
    for key, value in original.items():
        if value is not omit:
            new[key] = value
    return new


def maybe_filter_request_body(
    data: typing.Optional[typing.Any],
    request_options: typing.Optional[RequestOptions],
    omit: typing.Optional[typing.Any],
) -> typing.Optional[typing.Any]:
    if data is None:
        return (
            jsonable_encoder(request_options.get("additional_body_parameters", {})) or {}
            if request_options is not None
            else None
        )
    elif not isinstance(data, typing.Mapping):
        data_content = jsonable_encoder(data)
    else:
        data_content = {
            **(jsonable_encoder(remove_omit_from_dict(data, omit))),  # type: ignore
            **(
                jsonable_encoder(request_options.get("additional_body_parameters", {})) or {}
                if request_options is not None
                else {}
            ),
        }
    return data_content


# Abstracted out for testing purposes
def get_request_body(
    *,
    json: typing.Optional[typing.Any],
    data: typing.Optional[typing.Any],
    request_options: typing.Optional[RequestOptions],
    omit: typing.Optional[typing.Any],
) -> typing.Tuple[typing.Optional[typing.Any], typing.Optional[typing.Any]]:
    json_body = None
    data_body = None
    if data is not None:
        data_body = maybe_filter_request_body(data, request_options, omit)
    else:
        # If both data and json are None, we send json data in the event extra properties are specified
        json_body = maybe_filter_request_body(json, request_options, omit)

    has_additional_body_parameters = bool(
        request_options is not None and request_options.get("additional_body_parameters")
    )

    # Only collapse empty dict to None when the body was not explicitly provided
    # and there are no additional body parameters. This preserves explicit empty
    # bodies (e.g., when an endpoint has a request body type but all fields are optional).
    if json_body == {} and json is None and not has_additional_body_parameters:
        json_body = None
    if data_body == {} and data is None and not has_additional_body_parameters:
        data_body = None

    return json_body, data_body


class HttpClient:
    def __init__(
        self,
        *,
        httpx_client: httpx.Client,
        base_timeout: typing.Callable[[], typing.Optional[float]],
        base_headers: typing.Callable[[], typing.Dict[str, str]],
        base_url: typing.Optional[typing.Callable[[], str]] = None,
        base_max_retries: int = 2,
        logging_config: typing.Optional[typing.Union[LogConfig, Logger]] = None,
    ):
        self.base_url = base_url
        self.base_timeout = base_timeout
        self.base_headers = base_headers
        self.base_max_retries = base_max_retries
        self.httpx_client = httpx_client
        self.logger = create_logger(logging_config)

    def get_base_url(self, maybe_base_url: typing.Optional[str]) -> str:
        base_url = maybe_base_url
        if self.base_url is not None and base_url is None:
            base_url = self.base_url()

        if base_url is None:
            raise ValueError("A base_url is required to make this request, please provide one and try again.")
        return base_url

    def request(
        self,
        path: typing.Optional[str] = None,
        *,
        method: str,
        base_url: typing.Optional[str] = None,
        params: typing.Optional[typing.Dict[str, typing.Any]] = None,
        json: typing.Optional[typing.Any] = None,
        data: typing.Optional[typing.Any] = None,
        content: typing.Optional[typing.Union[bytes, typing.Iterator[bytes], typing.AsyncIterator[bytes]]] = None,
        files: typing.Optional[
            typing.Union[
                typing.Dict[str, typing.Optional[typing.Union[File, typing.List[File]]]],
                typing.List[typing.Tuple[str, File]],
            ]
        ] = None,
        headers: typing.Optional[typing.Dict[str, typing.Any]] = None,
        request_options: typing.Optional[RequestOptions] = None,
        retries: int = 0,
        omit: typing.Optional[typing.Any] = None,
        force_multipart: typing.Optional[bool] = None,
    ) -> httpx.Response:
        base_url = self.get_base_url(base_url)
        _timeout = (
            request_options.get("timeout")
            if request_options is not None and request_options.get("timeout") is not None
            else request_options.get("timeout_in_seconds")
            if request_options is not None and request_options.get("timeout_in_seconds") is not None
            else self.base_timeout()
        )
        timeout = _timeout if _timeout is not None else httpx.USE_CLIENT_DEFAULT

        json_body, data_body = get_request_body(json=json, data=data, request_options=request_options, omit=omit)

        request_files: typing.Optional[RequestFiles] = (
            convert_file_dict_to_httpx_tuples(remove_omit_from_dict(remove_none_from_dict(files), omit))
            if (files is not None and files is not omit and isinstance(files, dict))
            else None
        )

        if (request_files is None or len(request_files) == 0) and force_multipart:
            request_files = FORCE_MULTIPART

        data_body = _maybe_filter_none_from_multipart_data(data_body, request_files, force_multipart)

        # Compute encoded params separately to avoid passing empty list to httpx
        # (httpx strips existing query params from URL when params=[] is passed)
        _encoded_params = encode_query(
            jsonable_encoder(
                remove_none_from_dict(
                    remove_omit_from_dict(
                        {
                            **(params if params is not None else {}),
                            **(
                                request_options.get("additional_query_parameters", {}) or {}
                                if request_options is not None
                                else {}
                            ),
                        },
                        omit,
                    )
                )
            )
        )

        _request_url = _build_url(base_url, path)
        _request_headers = jsonable_encoder(
            remove_none_from_dict(
                {
                    **self.base_headers(),
                    **(headers if headers is not None else {}),
                    **(request_options.get("additional_headers", {}) or {} if request_options is not None else {}),
                }
            )
        )

        if self.logger.is_debug():
            self.logger.debug(
                "Making HTTP request",
                method=method,
                url=_request_url,
                headers=_redact_headers(_request_headers),
                has_body=json_body is not None or data_body is not None,
            )

        max_retries: int = (
            request_options.get("max_retries", self.base_max_retries)
            if request_options is not None
            else self.base_max_retries
        )

        try:
            response = self.httpx_client.request(
                method=method,
                url=_request_url,
                headers=_request_headers,
                params=_encoded_params if _encoded_params else None,
                json=json_body,
                data=data_body,
                content=content,
                files=request_files,
                timeout=timeout,
            )
        except (httpx.ConnectError, httpx.RemoteProtocolError):
            if retries < max_retries:
                time.sleep(_retry_timeout_from_retries(retries=retries))
                return self.request(
                    path=path,
                    method=method,
                    base_url=base_url,
                    params=params,
                    json=json,
                    data=data,
                    content=content,
                    files=files,
                    headers=headers,
                    request_options=request_options,
                    retries=retries + 1,
                    omit=omit,
                    force_multipart=force_multipart,
                )
            raise

        if _should_retry(response=response):
            if retries < max_retries:
                time.sleep(_retry_timeout(response=response, retries=retries))
                return self.request(
                    path=path,
                    method=method,
                    base_url=base_url,
                    params=params,
                    json=json,
                    data=data,
                    content=content,
                    files=files,
                    headers=headers,
                    request_options=request_options,
                    retries=retries + 1,
                    omit=omit,
                    force_multipart=force_multipart,
                )

        if self.logger.is_debug():
            if 200 <= response.status_code < 400:
                self.logger.debug(
                    "HTTP request succeeded",
                    method=method,
                    url=_request_url,
                    status_code=response.status_code,
                )

        if self.logger.is_error():
            if response.status_code >= 400:
                self.logger.error(
                    "HTTP request failed with error status",
                    method=method,
                    url=_request_url,
                    status_code=response.status_code,
                )

        return response

    @contextmanager
    def stream(
        self,
        path: typing.Optional[str] = None,
        *,
        method: str,
        base_url: typing.Optional[str] = None,
        params: typing.Optional[typing.Dict[str, typing.Any]] = None,
        json: typing.Optional[typing.Any] = None,
        data: typing.Optional[typing.Any] = None,
        content: typing.Optional[typing.Union[bytes, typing.Iterator[bytes], typing.AsyncIterator[bytes]]] = None,
        files: typing.Optional[
            typing.Union[
                typing.Dict[str, typing.Optional[typing.Union[File, typing.List[File]]]],
                typing.List[typing.Tuple[str, File]],
            ]
        ] = None,
        headers: typing.Optional[typing.Dict[str, typing.Any]] = None,
        request_options: typing.Optional[RequestOptions] = None,
        retries: int = 0,
        omit: typing.Optional[typing.Any] = None,
        force_multipart: typing.Optional[bool] = None,
    ) -> typing.Iterator[httpx.Response]:
        base_url = self.get_base_url(base_url)
        _timeout = (
            request_options.get("timeout")
            if request_options is not None and request_options.get("timeout") is not None
            else request_options.get("timeout_in_seconds")
            if request_options is not None and request_options.get("timeout_in_seconds") is not None
            else self.base_timeout()
        )
        timeout = _timeout if _timeout is not None else httpx.USE_CLIENT_DEFAULT

        request_files: typing.Optional[RequestFiles] = (
            convert_file_dict_to_httpx_tuples(remove_omit_from_dict(remove_none_from_dict(files), omit))
            if (files is not None and files is not omit and isinstance(files, dict))
            else None
        )

        if (request_files is None or len(request_files) == 0) and force_multipart:
            request_files = FORCE_MULTIPART

        json_body, data_body = get_request_body(json=json, data=data, request_options=request_options, omit=omit)

        data_body = _maybe_filter_none_from_multipart_data(data_body, request_files, force_multipart)

        # Compute encoded params separately to avoid passing empty list to httpx
        # (httpx strips existing query params from URL when params=[] is passed)
        _encoded_params = encode_query(
            jsonable_encoder(
                remove_none_from_dict(
                    remove_omit_from_dict(
                        {
                            **(params if params is not None else {}),
                            **(
                                request_options.get("additional_query_parameters", {})
                                if request_options is not None
                                else {}
                            ),
                        },
                        omit,
                    )
                )
            )
        )

        _request_url = _build_url(base_url, path)
        _request_headers = jsonable_encoder(
            remove_none_from_dict(
                {
                    **self.base_headers(),
                    **(headers if headers is not None else {}),
                    **(request_options.get("additional_headers", {}) if request_options is not None else {}),
                }
            )
        )

        if self.logger.is_debug():
            self.logger.debug(
                "Making streaming HTTP request",
                method=method,
                url=_request_url,
                headers=_redact_headers(_request_headers),
            )

        with self.httpx_client.stream(
            method=method,
            url=_request_url,
            headers=_request_headers,
            params=_encoded_params if _encoded_params else None,
            json=json_body,
            data=data_body,
            content=content,
            files=request_files,
            timeout=timeout,
        ) as stream:
            yield stream


class AsyncHttpClient:
    def __init__(
        self,
        *,
        httpx_client: httpx.AsyncClient,
        base_timeout: typing.Callable[[], typing.Optional[float]],
        base_headers: typing.Callable[[], typing.Dict[str, str]],
        base_url: typing.Optional[typing.Callable[[], str]] = None,
        base_max_retries: int = 2,
        async_base_headers: typing.Optional[typing.Callable[[], typing.Awaitable[typing.Dict[str, str]]]] = None,
        logging_config: typing.Optional[typing.Union[LogConfig, Logger]] = None,
    ):
        self.base_url = base_url
        self.base_timeout = base_timeout
        self.base_headers = base_headers
        self.base_max_retries = base_max_retries
        self.async_base_headers = async_base_headers
        self.httpx_client = httpx_client
        self.logger = create_logger(logging_config)

    async def _get_headers(self) -> typing.Dict[str, str]:
        if self.async_base_headers is not None:
            return await self.async_base_headers()
        return self.base_headers()

    def get_base_url(self, maybe_base_url: typing.Optional[str]) -> str:
        base_url = maybe_base_url
        if self.base_url is not None and base_url is None:
            base_url = self.base_url()

        if base_url is None:
            raise ValueError("A base_url is required to make this request, please provide one and try again.")
        return base_url

    async def request(
        self,
        path: typing.Optional[str] = None,
        *,
        method: str,
        base_url: typing.Optional[str] = None,
        params: typing.Optional[typing.Dict[str, typing.Any]] = None,
        json: typing.Optional[typing.Any] = None,
        data: typing.Optional[typing.Any] = None,
        content: typing.Optional[typing.Union[bytes, typing.Iterator[bytes], typing.AsyncIterator[bytes]]] = None,
        files: typing.Optional[
            typing.Union[
                typing.Dict[str, typing.Optional[typing.Union[File, typing.List[File]]]],
                typing.List[typing.Tuple[str, File]],
            ]
        ] = None,
        headers: typing.Optional[typing.Dict[str, typing.Any]] = None,
        request_options: typing.Optional[RequestOptions] = None,
        retries: int = 0,
        omit: typing.Optional[typing.Any] = None,
        force_multipart: typing.Optional[bool] = None,
    ) -> httpx.Response:
        base_url = self.get_base_url(base_url)
        _timeout = (
            request_options.get("timeout")
            if request_options is not None and request_options.get("timeout") is not None
            else request_options.get("timeout_in_seconds")
            if request_options is not None and request_options.get("timeout_in_seconds") is not None
            else self.base_timeout()
        )
        timeout = _timeout if _timeout is not None else httpx.USE_CLIENT_DEFAULT

        request_files: typing.Optional[RequestFiles] = (
            convert_file_dict_to_httpx_tuples(remove_omit_from_dict(remove_none_from_dict(files), omit))
            if (files is not None and files is not omit and isinstance(files, dict))
            else None
        )

        if (request_files is None or len(request_files) == 0) and force_multipart:
            request_files = FORCE_MULTIPART

        json_body, data_body = get_request_body(json=json, data=data, request_options=request_options, omit=omit)

        data_body = _maybe_filter_none_from_multipart_data(data_body, request_files, force_multipart)

        # Get headers (supports async token providers)
        _headers = await self._get_headers()

        # Compute encoded params separately to avoid passing empty list to httpx
        # (httpx strips existing query params from URL when params=[] is passed)
        _encoded_params = encode_query(
            jsonable_encoder(
                remove_none_from_dict(
                    remove_omit_from_dict(
                        {
                            **(params if params is not None else {}),
                            **(
                                request_options.get("additional_query_parameters", {}) or {}
                                if request_options is not None
                                else {}
                            ),
                        },
                        omit,
                    )
                )
            )
        )

        _request_url = _build_url(base_url, path)
        _request_headers = jsonable_encoder(
            remove_none_from_dict(
                {
                    **_headers,
                    **(headers if headers is not None else {}),
                    **(request_options.get("additional_headers", {}) or {} if request_options is not None else {}),
                }
            )
        )

        if self.logger.is_debug():
            self.logger.debug(
                "Making HTTP request",
                method=method,
                url=_request_url,
                headers=_redact_headers(_request_headers),
                has_body=json_body is not None or data_body is not None,
            )

        max_retries: int = (
            request_options.get("max_retries", self.base_max_retries)
            if request_options is not None
            else self.base_max_retries
        )

        try:
            response = await self.httpx_client.request(
                method=method,
                url=_request_url,
                headers=_request_headers,
                params=_encoded_params if _encoded_params else None,
                json=json_body,
                data=data_body,
                content=content,
                files=request_files,
                timeout=timeout,
            )
        except (httpx.ConnectError, httpx.RemoteProtocolError):
            if retries < max_retries:
                await asyncio.sleep(_retry_timeout_from_retries(retries=retries))
                return await self.request(
                    path=path,
                    method=method,
                    base_url=base_url,
                    params=params,
                    json=json,
                    data=data,
                    content=content,
                    files=files,
                    headers=headers,
                    request_options=request_options,
                    retries=retries + 1,
                    omit=omit,
                    force_multipart=force_multipart,
                )
            raise

        if _should_retry(response=response):
            if retries < max_retries:
                await asyncio.sleep(_retry_timeout(response=response, retries=retries))
                return await self.request(
                    path=path,
                    method=method,
                    base_url=base_url,
                    params=params,
                    json=json,
                    data=data,
                    content=content,
                    files=files,
                    headers=headers,
                    request_options=request_options,
                    retries=retries + 1,
                    omit=omit,
                    force_multipart=force_multipart,
                )

        if self.logger.is_debug():
            if 200 <= response.status_code < 400:
                self.logger.debug(
                    "HTTP request succeeded",
                    method=method,
                    url=_request_url,
                    status_code=response.status_code,
                )

        if self.logger.is_error():
            if response.status_code >= 400:
                self.logger.error(
                    "HTTP request failed with error status",
                    method=method,
                    url=_request_url,
                    status_code=response.status_code,
                )

        return response

    @asynccontextmanager
    async def stream(
        self,
        path: typing.Optional[str] = None,
        *,
        method: str,
        base_url: typing.Optional[str] = None,
        params: typing.Optional[typing.Dict[str, typing.Any]] = None,
        json: typing.Optional[typing.Any] = None,
        data: typing.Optional[typing.Any] = None,
        content: typing.Optional[typing.Union[bytes, typing.Iterator[bytes], typing.AsyncIterator[bytes]]] = None,
        files: typing.Optional[
            typing.Union[
                typing.Dict[str, typing.Optional[typing.Union[File, typing.List[File]]]],
                typing.List[typing.Tuple[str, File]],
            ]
        ] = None,
        headers: typing.Optional[typing.Dict[str, typing.Any]] = None,
        request_options: typing.Optional[RequestOptions] = None,
        retries: int = 0,
        omit: typing.Optional[typing.Any] = None,
        force_multipart: typing.Optional[bool] = None,
    ) -> typing.AsyncIterator[httpx.Response]:
        base_url = self.get_base_url(base_url)
        _timeout = (
            request_options.get("timeout")
            if request_options is not None and request_options.get("timeout") is not None
            else request_options.get("timeout_in_seconds")
            if request_options is not None and request_options.get("timeout_in_seconds") is not None
            else self.base_timeout()
        )
        timeout = _timeout if _timeout is not None else httpx.USE_CLIENT_DEFAULT

        request_files: typing.Optional[RequestFiles] = (
            convert_file_dict_to_httpx_tuples(remove_omit_from_dict(remove_none_from_dict(files), omit))
            if (files is not None and files is not omit and isinstance(files, dict))
            else None
        )

        if (request_files is None or len(request_files) == 0) and force_multipart:
            request_files = FORCE_MULTIPART

        json_body, data_body = get_request_body(json=json, data=data, request_options=request_options, omit=omit)

        data_body = _maybe_filter_none_from_multipart_data(data_body, request_files, force_multipart)

        # Get headers (supports async token providers)
        _headers = await self._get_headers()

        # Compute encoded params separately to avoid passing empty list to httpx
        # (httpx strips existing query params from URL when params=[] is passed)
        _encoded_params = encode_query(
            jsonable_encoder(
                remove_none_from_dict(
                    remove_omit_from_dict(
                        {
                            **(params if params is not None else {}),
                            **(
                                request_options.get("additional_query_parameters", {})
                                if request_options is not None
                                else {}
                            ),
                        },
                        omit=omit,
                    )
                )
            )
        )

        _request_url = _build_url(base_url, path)
        _request_headers = jsonable_encoder(
            remove_none_from_dict(
                {
                    **_headers,
                    **(headers if headers is not None else {}),
                    **(request_options.get("additional_headers", {}) if request_options is not None else {}),
                }
            )
        )

        if self.logger.is_debug():
            self.logger.debug(
                "Making streaming HTTP request",
                method=method,
                url=_request_url,
                headers=_redact_headers(_request_headers),
            )

        async with self.httpx_client.stream(
            method=method,
            url=_request_url,
            headers=_request_headers,
            params=_encoded_params if _encoded_params else None,
            json=json_body,
            data=data_body,
            content=content,
            files=request_files,
            timeout=timeout,
        ) as stream:
            yield stream