from __future__ import annotations
import contextlib
-import ssl
import typing
from types import TracebackType
-import httpcore
+if typing.TYPE_CHECKING:
+ import ssl # pragma: no cover
+
+ import httpx # pragma: no cover
from .._config import DEFAULT_LIMITS, Limits, Proxy, SSLContext, create_ssl_context
from .._exceptions import (
__all__ = ["AsyncHTTPTransport", "HTTPTransport"]
+HTTPCORE_EXC_MAP: dict[type[Exception], type[httpx.HTTPError]] = {}
+
+
+def _load_httpcore_exceptions() -> dict[type[Exception], type[httpx.HTTPError]]:
+ import httpcore
+
+ return {
+ httpcore.TimeoutException: TimeoutException,
+ httpcore.ConnectTimeout: ConnectTimeout,
+ httpcore.ReadTimeout: ReadTimeout,
+ httpcore.WriteTimeout: WriteTimeout,
+ httpcore.PoolTimeout: PoolTimeout,
+ httpcore.NetworkError: NetworkError,
+ httpcore.ConnectError: ConnectError,
+ httpcore.ReadError: ReadError,
+ httpcore.WriteError: WriteError,
+ httpcore.ProxyError: ProxyError,
+ httpcore.UnsupportedProtocol: UnsupportedProtocol,
+ httpcore.ProtocolError: ProtocolError,
+ httpcore.LocalProtocolError: LocalProtocolError,
+ httpcore.RemoteProtocolError: RemoteProtocolError,
+ }
+
@contextlib.contextmanager
def map_httpcore_exceptions() -> typing.Iterator[None]:
+ global HTTPCORE_EXC_MAP
+ if len(HTTPCORE_EXC_MAP) == 0:
+ HTTPCORE_EXC_MAP = _load_httpcore_exceptions()
try:
yield
except Exception as exc:
raise mapped_exc(message) from exc
-HTTPCORE_EXC_MAP = {
- httpcore.TimeoutException: TimeoutException,
- httpcore.ConnectTimeout: ConnectTimeout,
- httpcore.ReadTimeout: ReadTimeout,
- httpcore.WriteTimeout: WriteTimeout,
- httpcore.PoolTimeout: PoolTimeout,
- httpcore.NetworkError: NetworkError,
- httpcore.ConnectError: ConnectError,
- httpcore.ReadError: ReadError,
- httpcore.WriteError: WriteError,
- httpcore.ProxyError: ProxyError,
- httpcore.UnsupportedProtocol: UnsupportedProtocol,
- httpcore.ProtocolError: ProtocolError,
- httpcore.LocalProtocolError: LocalProtocolError,
- httpcore.RemoteProtocolError: RemoteProtocolError,
-}
-
-
class ResponseStream(SyncByteStream):
def __init__(self, httpcore_stream: typing.Iterable[bytes]) -> None:
self._httpcore_stream = httpcore_stream
verify: typing.Any = None,
cert: typing.Any = None,
) -> None:
+ import httpcore
+
proxy = Proxy(url=proxy) if isinstance(proxy, (str, URL)) else proxy
if verify is not None or cert is not None: # pragma: nocover
# Deprecated...
request: Request,
) -> Response:
assert isinstance(request.stream, SyncByteStream)
+ import httpcore
req = httpcore.Request(
method=request.method,
verify: typing.Any = None,
cert: typing.Any = None,
) -> None:
+ import httpcore
+
proxy = Proxy(url=proxy) if isinstance(proxy, (str, URL)) else proxy
if verify is not None or cert is not None: # pragma: nocover
# Deprecated...
request: Request,
) -> Response:
assert isinstance(request.stream, AsyncByteStream)
+ import httpcore
req = httpcore.Request(
method=request.method,