### Usage
For some advanced configuration you might need to instantiate a transport
-class directly, and pass it to the client instance. The `httpcore` package
-provides a `local_address` configuration that is only available via this
-low-level API.
+class directly, and pass it to the client instance. One example is the
+`local_address` configuration which is only available via this low-level API.
```pycon
->>> import httpx, httpcore
->>> ssl_context = httpx.create_ssl_context()
->>> transport = httpcore.SyncConnectionPool(
-... ssl_context=ssl_context,
-... max_connections=100,
-... max_keepalive_connections=20,
-... keepalive_expiry=5.0,
-... local_address="0.0.0.0"
-... ) # Use the standard HTTPX defaults, but with an IPv4 only 'local_address'.
+>>> import httpx
+>>> transport = httpx.HTTPTransport(local_address="0.0.0.0")
>>> client = httpx.Client(transport=transport)
```
-Similarly, `httpcore` provides a `uds` option for connecting via a Unix Domain Socket that is only available via this low-level API:
+Connection retries are also available via this interface.
-```python
->>> import httpx, httpcore
->>> ssl_context = httpx.create_ssl_context()
->>> transport = httpcore.SyncConnectionPool(
-... ssl_context=ssl_context,
-... max_connections=100,
-... max_keepalive_connections=20,
-... keepalive_expiry=5.0,
-... uds="/var/run/docker.sock",
-... ) # Connect to the Docker API via a Unix Socket.
+```pycon
+>>> import httpx
+>>> transport = httpx.HTTPTransport(retries=1)
+>>> client = httpx.Client(transport=transport)
+```
+
+Similarly, instantiating a transport directly provides a `uds` option for
+connecting via a Unix Domain Socket that is only available via this low-level API:
+
+```pycon
+>>> import httpx
+>>> # Connect to the Docker API via a Unix Socket.
+>>> transport = httpx.HTTPTransport(uds="/var/run/docker.sock")
>>> client = httpx.Client(transport=transport)
>>> response = client.get("http://docker/info")
>>> response.json()
{"ID": "...", "Containers": 4, "Images": 74, ...}
```
-Unlike the `httpx.Client()`, the lower-level `httpcore` transport instances
-do not include any default values for configuring aspects such as the
-connection pooling details, so you'll need to provide more explicit
-configuration when using this API.
-
### urllib3 transport
This [public gist](https://gist.github.com/florimondmanca/d56764d78d748eb9f73165da388e546e) provides a transport that uses the excellent [`urllib3` library](https://urllib3.readthedocs.io/en/latest/), and can be used with the sync `Client`...
A couple of other sketches of how you might take advantage of mounted transports...
+Disabling HTTP/2 on a single given domain...
+
+```python
+mounts = {
+ "all://": httpx.HTTPTransport(http2=True),
+ "all://*example.org": httpx.HTTPTransport()
+}
+client = httpx.Client(mounts=mounts)
+```
+
Mocking requests to a given domain:
```python
await client.post(url, data=upload_bytes())
```
+### Explicit transport instances
+
+When instantiating a transport instance directly, you need to use `httpx.AsyncHTTPTransport`.
+
+For instance:
+
+```pycon
+>>> import httpx
+>>> transport = httpx.AsyncHTTPTransport(retries=1)
+>>> async with httpx.AsyncClient(transport=transport) as client:
+>>> ...
+```
+
## Supported async environments
HTTPX supports either `asyncio` or `trio` as an async environment.
from ._models import URL, Cookies, Headers, QueryParams, Request, Response
from ._status_codes import StatusCode, codes
from ._transports.asgi import ASGITransport
+from ._transports.default import AsyncHTTPTransport, HTTPTransport
from ._transports.mock import MockTransport
from ._transports.wsgi import WSGITransport
"__version__",
"ASGITransport",
"AsyncClient",
+ "AsyncHTTPTransport",
"Auth",
"BasicAuth",
"Client",
"Headers",
"HTTPError",
"HTTPStatusError",
+ "HTTPTransport",
"InvalidURL",
"Limits",
"LocalProtocolError",
Proxy,
Timeout,
UnsetType,
- create_ssl_context,
)
from ._decoders import SUPPORTED_DECODERS
from ._exceptions import (
from ._models import URL, Cookies, Headers, QueryParams, Request, Response
from ._status_codes import codes
from ._transports.asgi import ASGITransport
+from ._transports.default import AsyncHTTPTransport, HTTPTransport
from ._transports.wsgi import WSGITransport
from ._types import (
AuthTypes,
if app is not None:
return WSGITransport(app=app)
- ssl_context = create_ssl_context(verify=verify, cert=cert, trust_env=trust_env)
-
- return httpcore.SyncConnectionPool(
- ssl_context=ssl_context,
- max_connections=limits.max_connections,
- max_keepalive_connections=limits.max_keepalive_connections,
- keepalive_expiry=limits.keepalive_expiry,
- http2=http2,
+ return HTTPTransport(
+ verify=verify, cert=cert, http2=http2, limits=limits, trust_env=trust_env
)
def _init_proxy_transport(
limits: Limits = DEFAULT_LIMITS,
trust_env: bool = True,
) -> httpcore.SyncHTTPTransport:
- ssl_context = create_ssl_context(verify=verify, cert=cert, trust_env=trust_env)
-
- return httpcore.SyncHTTPProxy(
- proxy_url=proxy.url.raw,
- proxy_headers=proxy.headers.raw,
- proxy_mode=proxy.mode,
- ssl_context=ssl_context,
- max_connections=limits.max_connections,
- max_keepalive_connections=limits.max_keepalive_connections,
- keepalive_expiry=limits.keepalive_expiry,
+ return HTTPTransport(
+ verify=verify,
+ cert=cert,
http2=http2,
+ limits=limits,
+ trust_env=trust_env,
+ proxy=proxy,
)
def _transport_for_url(self, url: URL) -> httpcore.SyncHTTPTransport:
if app is not None:
return ASGITransport(app=app)
- ssl_context = create_ssl_context(verify=verify, cert=cert, trust_env=trust_env)
-
- return httpcore.AsyncConnectionPool(
- ssl_context=ssl_context,
- max_connections=limits.max_connections,
- max_keepalive_connections=limits.max_keepalive_connections,
- keepalive_expiry=limits.keepalive_expiry,
- http2=http2,
+ return AsyncHTTPTransport(
+ verify=verify, cert=cert, http2=http2, limits=limits, trust_env=trust_env
)
def _init_proxy_transport(
limits: Limits = DEFAULT_LIMITS,
trust_env: bool = True,
) -> httpcore.AsyncHTTPTransport:
- ssl_context = create_ssl_context(verify=verify, cert=cert, trust_env=trust_env)
-
- return httpcore.AsyncHTTPProxy(
- proxy_url=proxy.url.raw,
- proxy_headers=proxy.headers.raw,
- proxy_mode=proxy.mode,
- ssl_context=ssl_context,
- max_connections=limits.max_connections,
- max_keepalive_connections=limits.max_keepalive_connections,
- keepalive_expiry=limits.keepalive_expiry,
+ return AsyncHTTPTransport(
+ verify=verify,
+ cert=cert,
http2=http2,
+ limits=limits,
+ trust_env=trust_env,
+ proxy=proxy,
)
def _transport_for_url(self, url: URL) -> httpcore.AsyncHTTPTransport:
--- /dev/null
+"""
+Custom transports, with nicely configured defaults.
+
+The following additional keyword arguments are currently supported by httpcore...
+
+* uds: str
+* local_address: str
+* retries: int
+* backend: str ("auto", "asyncio", "trio", "curio", "anyio", "sync")
+
+Example usages...
+
+# Disable HTTP/2 on a single specfic domain.
+mounts = {
+ "all://": httpx.HTTPTransport(http2=True),
+ "all://*example.org": httpx.HTTPTransport()
+}
+
+# Using advanced httpcore configuration, with connection retries.
+transport = httpx.HTTPTransport(retries=1)
+client = httpx.Client(transport=transport)
+
+# Using advanced httpcore configuration, with unix domain sockets.
+transport = httpx.HTTPTransport(uds="socket.uds")
+client = httpx.Client(transport=transport)
+"""
+import typing
+from types import TracebackType
+
+import httpcore
+
+from .._config import DEFAULT_LIMITS, Limits, Proxy, create_ssl_context
+from .._types import CertTypes, VerifyTypes
+
+T = typing.TypeVar("T", bound="HTTPTransport")
+A = typing.TypeVar("A", bound="AsyncHTTPTransport")
+Headers = typing.List[typing.Tuple[bytes, bytes]]
+URL = typing.Tuple[bytes, bytes, typing.Optional[int], bytes]
+
+
+class HTTPTransport(httpcore.SyncHTTPTransport):
+ def __init__(
+ self,
+ verify: VerifyTypes = True,
+ cert: CertTypes = None,
+ http2: bool = False,
+ limits: Limits = DEFAULT_LIMITS,
+ trust_env: bool = True,
+ proxy: Proxy = None,
+ uds: str = None,
+ local_address: str = None,
+ retries: int = 0,
+ backend: str = "sync",
+ ) -> None:
+ ssl_context = create_ssl_context(verify=verify, cert=cert, trust_env=trust_env)
+
+ if proxy is None:
+ self._pool = httpcore.SyncConnectionPool(
+ ssl_context=ssl_context,
+ max_connections=limits.max_connections,
+ max_keepalive_connections=limits.max_keepalive_connections,
+ keepalive_expiry=limits.keepalive_expiry,
+ http2=http2,
+ uds=uds,
+ local_address=local_address,
+ retries=retries,
+ backend=backend,
+ )
+ else:
+ self._pool = httpcore.SyncHTTPProxy(
+ proxy_url=proxy.url.raw,
+ proxy_headers=proxy.headers.raw,
+ proxy_mode=proxy.mode,
+ ssl_context=ssl_context,
+ max_connections=limits.max_connections,
+ max_keepalive_connections=limits.max_keepalive_connections,
+ keepalive_expiry=limits.keepalive_expiry,
+ http2=http2,
+ backend=backend,
+ )
+
+ def __enter__(self: T) -> T: # Use generics for subclass support.
+ self._pool.__enter__()
+ return self
+
+ def __exit__(
+ self,
+ exc_type: typing.Type[BaseException] = None,
+ exc_value: BaseException = None,
+ traceback: TracebackType = None,
+ ) -> None:
+ self._pool.__exit__(exc_type, exc_value, traceback)
+
+ def request(
+ self,
+ method: bytes,
+ url: URL,
+ headers: Headers = None,
+ stream: httpcore.SyncByteStream = None,
+ ext: dict = None,
+ ) -> typing.Tuple[int, Headers, httpcore.SyncByteStream, dict]:
+ return self._pool.request(method, url, headers=headers, stream=stream, ext=ext)
+
+ def close(self) -> None:
+ self._pool.close()
+
+
+class AsyncHTTPTransport(httpcore.AsyncHTTPTransport):
+ def __init__(
+ self,
+ verify: VerifyTypes = True,
+ cert: CertTypes = None,
+ http2: bool = False,
+ limits: Limits = DEFAULT_LIMITS,
+ trust_env: bool = True,
+ proxy: Proxy = None,
+ uds: str = None,
+ local_address: str = None,
+ retries: int = 0,
+ backend: str = "auto",
+ ) -> None:
+ ssl_context = create_ssl_context(verify=verify, cert=cert, trust_env=trust_env)
+
+ if proxy is None:
+ self._pool = httpcore.AsyncConnectionPool(
+ ssl_context=ssl_context,
+ max_connections=limits.max_connections,
+ max_keepalive_connections=limits.max_keepalive_connections,
+ keepalive_expiry=limits.keepalive_expiry,
+ http2=http2,
+ uds=uds,
+ local_address=local_address,
+ retries=retries,
+ backend=backend,
+ )
+ else:
+ self._pool = httpcore.AsyncHTTPProxy(
+ proxy_url=proxy.url.raw,
+ proxy_headers=proxy.headers.raw,
+ proxy_mode=proxy.mode,
+ ssl_context=ssl_context,
+ max_connections=limits.max_connections,
+ max_keepalive_connections=limits.max_keepalive_connections,
+ keepalive_expiry=limits.keepalive_expiry,
+ http2=http2,
+ backend=backend,
+ )
+
+ async def __aenter__(self: A) -> A: # Use generics for subclass support.
+ await self._pool.__aenter__()
+ return self
+
+ async def __aexit__(
+ self,
+ exc_type: typing.Type[BaseException] = None,
+ exc_value: BaseException = None,
+ traceback: TracebackType = None,
+ ) -> None:
+ await self._pool.__aexit__(exc_type, exc_value, traceback)
+
+ async def arequest(
+ self,
+ method: bytes,
+ url: URL,
+ headers: Headers = None,
+ stream: httpcore.AsyncByteStream = None,
+ ext: dict = None,
+ ) -> typing.Tuple[int, Headers, httpcore.AsyncByteStream, dict]:
+ return await self._pool.arequest(
+ method, url, headers=headers, stream=stream, ext=ext
+ )
+
+ async def aclose(self) -> None:
+ await self._pool.aclose()
pattern = URLPattern(proxy_key)
assert pattern in client._mounts
proxy = client._mounts[pattern]
- assert isinstance(proxy, httpcore.SyncHTTPProxy)
- assert proxy.proxy_origin == url_to_origin(url)
+ assert isinstance(proxy, httpx.HTTPTransport)
+ assert isinstance(proxy._pool, httpcore.SyncHTTPProxy)
+ assert proxy._pool.proxy_origin == url_to_origin(url)
assert len(expected_proxies) == len(client._mounts)
if expected is None:
assert transport is client._transport
else:
- assert isinstance(transport, httpcore.SyncHTTPProxy)
- assert transport.proxy_origin == url_to_origin(expected)
+ assert isinstance(transport, httpx.HTTPTransport)
+ assert isinstance(transport._pool, httpcore.SyncHTTPProxy)
+ assert transport._pool.proxy_origin == url_to_origin(expected)
@pytest.mark.asyncio
if expected is None:
assert transport == client._transport
else:
- assert transport.proxy_origin == url_to_origin(expected)
+ assert transport._pool.proxy_origin == url_to_origin(expected)
@pytest.mark.parametrize(
projects / thirdparty / httpx.git / commitdiff
