from ._api import delete, get, head, options, patch, post, put, request, stream
from ._auth import Auth, BasicAuth, DigestAuth, NetRCAuth
from ._client import USE_CLIENT_DEFAULT, AsyncClient, Client
-from ._config import Limits, Proxy, Timeout, create_ssl_context
+from ._config import Limits, Proxy, Timeout, Version, create_ssl_context
from ._content import ByteStream
from ._exceptions import (
CloseError,
"UnsupportedProtocol",
"URL",
"USE_CLIENT_DEFAULT",
+ "Version",
"WriteError",
"WriteTimeout",
"WSGITransport",
DEFAULT_LIMITS,
DEFAULT_MAX_REDIRECTS,
DEFAULT_TIMEOUT_CONFIG,
+ DEFAULT_VERSION,
Limits,
Proxy,
Timeout,
+ Version,
)
from ._decoders import SUPPORTED_DECODERS
from ._exceptions import (
cookies: typing.Optional[CookieTypes] = None,
verify: VerifyTypes = True,
cert: typing.Optional[CertTypes] = None,
- http1: bool = True,
http2: bool = False,
+ version: Version = DEFAULT_VERSION,
proxy: typing.Optional[ProxyTypes] = None,
proxies: typing.Optional[ProxiesTypes] = None,
mounts: typing.Optional[
default_encoding=default_encoding,
)
- if http2:
+ if "HTTP/2" in version:
try:
import h2 # noqa
except ImportError: # pragma: no cover
raise ImportError(
- "Using http2=True, but the 'h2' package is not installed. "
+ "Configured 'HTTP/2', but the 'h2' package is not installed. "
"Make sure to install httpx using `pip install httpx[http2]`."
) from None
if proxy:
raise RuntimeError("Use either `proxy` or 'proxies', not both.")
+ if http2:
+ raise RuntimeError(
+ "The 'http2' argument is now deprecated. "
+ "Use version=httpx.Version('HTTP/1.1', 'HTTP/2')."
+ )
+
allow_env_proxies = trust_env and app is None and transport is None
proxy_map = self._get_proxy_map(proxies or proxy, allow_env_proxies)
self._transport = self._init_transport(
verify=verify,
cert=cert,
- http1=http1,
- http2=http2,
+ version=version,
limits=limits,
transport=transport,
app=app,
proxy,
verify=verify,
cert=cert,
- http1=http1,
- http2=http2,
+ version=version,
limits=limits,
trust_env=trust_env,
)
self,
verify: VerifyTypes = True,
cert: typing.Optional[CertTypes] = None,
- http1: bool = True,
- http2: bool = False,
+ version: Version = DEFAULT_VERSION,
limits: Limits = DEFAULT_LIMITS,
transport: typing.Optional[BaseTransport] = None,
app: typing.Optional[typing.Callable[..., typing.Any]] = None,
return HTTPTransport(
verify=verify,
cert=cert,
- http1=http1,
- http2=http2,
+ version=version,
limits=limits,
trust_env=trust_env,
)
proxy: Proxy,
verify: VerifyTypes = True,
cert: typing.Optional[CertTypes] = None,
- http1: bool = True,
- http2: bool = False,
+ version: Version = DEFAULT_VERSION,
limits: Limits = DEFAULT_LIMITS,
trust_env: bool = True,
) -> BaseTransport:
return HTTPTransport(
verify=verify,
cert=cert,
- http1=http1,
- http2=http2,
+ version=version,
limits=limits,
trust_env=trust_env,
proxy=proxy,
cookies: typing.Optional[CookieTypes] = None,
verify: VerifyTypes = True,
cert: typing.Optional[CertTypes] = None,
- http1: bool = True,
http2: bool = False,
+ version: Version = DEFAULT_VERSION,
proxy: typing.Optional[ProxyTypes] = None,
proxies: typing.Optional[ProxiesTypes] = None,
mounts: typing.Optional[
default_encoding=default_encoding,
)
- if http2:
+ if "HTTP/2" in version:
try:
import h2 # noqa
except ImportError: # pragma: no cover
raise ImportError(
- "Using http2=True, but the 'h2' package is not installed. "
+ "Configured 'HTTP/2', but the 'h2' package is not installed. "
"Make sure to install httpx using `pip install httpx[http2]`."
) from None
if proxies:
message = (
- "The 'proxies' argument is now deprecated."
- " Use 'proxy' or 'mounts' instead."
+ "The 'proxies' argument is now deprecated. "
+ "Use 'proxy' or 'mounts' instead."
)
warnings.warn(message, DeprecationWarning)
if proxy:
raise RuntimeError("Use either `proxy` or 'proxies', not both.")
+ if http2:
+ raise RuntimeError(
+ "The 'http2' argument is now deprecated. "
+ "Use version=httpx.Version('HTTP/1.1', 'HTTP/2')."
+ )
+
allow_env_proxies = trust_env and app is None and transport is None
proxy_map = self._get_proxy_map(proxies or proxy, allow_env_proxies)
self._transport = self._init_transport(
verify=verify,
cert=cert,
- http1=http1,
- http2=http2,
+ version=version,
limits=limits,
transport=transport,
app=app,
proxy,
verify=verify,
cert=cert,
- http1=http1,
- http2=http2,
+ version=version,
limits=limits,
trust_env=trust_env,
)
self,
verify: VerifyTypes = True,
cert: typing.Optional[CertTypes] = None,
- http1: bool = True,
- http2: bool = False,
+ version: Version = DEFAULT_VERSION,
limits: Limits = DEFAULT_LIMITS,
transport: typing.Optional[AsyncBaseTransport] = None,
app: typing.Optional[typing.Callable[..., typing.Any]] = None,
return AsyncHTTPTransport(
verify=verify,
cert=cert,
- http1=http1,
- http2=http2,
+ version=version,
limits=limits,
trust_env=trust_env,
)
proxy: Proxy,
verify: VerifyTypes = True,
cert: typing.Optional[CertTypes] = None,
- http1: bool = True,
- http2: bool = False,
+ version: Version = DEFAULT_VERSION,
limits: Limits = DEFAULT_LIMITS,
trust_env: bool = True,
) -> AsyncBaseTransport:
return AsyncHTTPTransport(
verify=verify,
cert=cert,
- http1=http1,
- http2=http2,
+ version=version,
limits=limits,
trust_env=trust_env,
proxy=proxy,
return f"Proxy({url_str}{auth_str}{headers_str})"
+class Version:
+ def __init__(self, *versions: str) -> None:
+ self._versions = sorted(set(versions))
+ if any([version not in ["HTTP/1.1", "HTTP/2"] for version in versions]):
+ raise ValueError("Supported versions are 'HTTP/1.1' and 'HTTP/2'")
+
+ def __contains__(self, version: str) -> bool:
+ return version in self._versions
+
+ def __repr__(self) -> str:
+ version_str = ", ".join([repr(version) for version in self._versions])
+ return f"Version({version_str})"
+
+
DEFAULT_TIMEOUT_CONFIG = Timeout(timeout=5.0)
DEFAULT_LIMITS = Limits(max_connections=100, max_keepalive_connections=20)
DEFAULT_MAX_REDIRECTS = 20
+DEFAULT_VERSION = Version("HTTP/1.1")
Example usages...
-# Disable HTTP/2 on a single specific domain.
+# Enable HTTP/2, except on a single specific domain.
+
+ENABLE_HTTP2 = httpx.Version("HTTP/1.1", "HTTP/2")
mounts = {
- "all://": httpx.HTTPTransport(http2=True),
+ "all://": httpx.HTTPTransport(version=ENABLE_HTTP2),
"all://*example.org": httpx.HTTPTransport()
}
import httpcore
-from .._config import DEFAULT_LIMITS, Limits, Proxy, create_ssl_context
+from .._config import (
+ DEFAULT_LIMITS,
+ DEFAULT_VERSION,
+ Limits,
+ Proxy,
+ Version,
+ create_ssl_context,
+)
from .._exceptions import (
ConnectError,
ConnectTimeout,
self,
verify: VerifyTypes = True,
cert: typing.Optional[CertTypes] = None,
- http1: bool = True,
- http2: bool = False,
+ version: Version = DEFAULT_VERSION,
limits: Limits = DEFAULT_LIMITS,
trust_env: bool = True,
proxy: typing.Optional[ProxyTypes] = None,
max_connections=limits.max_connections,
max_keepalive_connections=limits.max_keepalive_connections,
keepalive_expiry=limits.keepalive_expiry,
- http1=http1,
- http2=http2,
+ http1="HTTP/1.1" in version,
+ http2="HTTP/2" in version,
uds=uds,
local_address=local_address,
retries=retries,
max_connections=limits.max_connections,
max_keepalive_connections=limits.max_keepalive_connections,
keepalive_expiry=limits.keepalive_expiry,
- http1=http1,
- http2=http2,
+ http1="HTTP/1.1" in version,
+ http2="HTTP/2" in version,
socket_options=socket_options,
)
elif proxy.url.scheme == "socks5":
max_connections=limits.max_connections,
max_keepalive_connections=limits.max_keepalive_connections,
keepalive_expiry=limits.keepalive_expiry,
- http1=http1,
- http2=http2,
+ http1="HTTP/1.1" in version,
+ http2="HTTP/2" in version,
)
else: # pragma: no cover
raise ValueError(
self,
verify: VerifyTypes = True,
cert: typing.Optional[CertTypes] = None,
- http1: bool = True,
- http2: bool = False,
+ version: Version = DEFAULT_VERSION,
limits: Limits = DEFAULT_LIMITS,
trust_env: bool = True,
proxy: typing.Optional[ProxyTypes] = None,
max_connections=limits.max_connections,
max_keepalive_connections=limits.max_keepalive_connections,
keepalive_expiry=limits.keepalive_expiry,
- http1=http1,
- http2=http2,
+ http1="HTTP/1.1" in version,
+ http2="HTTP/2" in version,
uds=uds,
local_address=local_address,
retries=retries,
max_connections=limits.max_connections,
max_keepalive_connections=limits.max_keepalive_connections,
keepalive_expiry=limits.keepalive_expiry,
- http1=http1,
- http2=http2,
+ http1="HTTP/1.1" in version,
+ http2="HTTP/2" in version,
socket_options=socket_options,
)
elif proxy.url.scheme == "socks5":
max_connections=limits.max_connections,
max_keepalive_connections=limits.max_keepalive_connections,
keepalive_expiry=limits.keepalive_expiry,
- http1=http1,
- http2=http2,
+ http1="HTTP/1.1" in version,
+ http2="HTTP/2" in version,
)
else: # pragma: no cover
raise ValueError(
@pytest.mark.anyio
async def test_get(server):
url = server.url
- async with httpx.AsyncClient(http2=True) as client:
+ ENABLE_HTTP2 = httpx.Version("HTTP/1.1", "HTTP/2")
+ async with httpx.AsyncClient(version=ENABLE_HTTP2) as client:
response = await client.get(url)
assert response.status_code == 200
assert response.text == "Hello, world!"
@pytest.mark.anyio
async def test_server_extensions(server):
url = server.url
- async with httpx.AsyncClient(http2=True) as client:
+ ENABLE_HTTP2 = httpx.Version("HTTP/1.1", "HTTP/2")
+ async with httpx.AsyncClient(version=ENABLE_HTTP2) as client:
response = await client.get(url)
assert response.status_code == 200
assert response.extensions["http_version"] == b"HTTP/1.1"
def test_get(server):
url = server.url
- with httpx.Client(http2=True) as http:
+ ENABLE_HTTP2 = httpx.Version("HTTP/1.1", "HTTP/2")
+ with httpx.Client(version=ENABLE_HTTP2) as http:
response = http.get(url)
assert response.status_code == 200
assert response.url == url
def test_server_extensions(server):
url = server.url.copy_with(path="/http_version_2")
- with httpx.Client(http2=True) as client:
+ ENABLE_HTTP2 = httpx.Version("HTTP/1.1", "HTTP/2")
+ with httpx.Client(version=ENABLE_HTTP2) as client:
response = client.get(url)
assert response.status_code == 200
assert response.extensions["http_version"] == b"HTTP/1.1"
def test_invalid_proxy_scheme():
with pytest.raises(ValueError):
httpx.Proxy("invalid://example.com")
+
+
+def test_version():
+ version = httpx.Version("HTTP/1.1", "HTTP/2")
+ assert "HTTP/1.1" in version
+ assert "HTTP/2" in version
+ assert repr(version) == "Version('HTTP/1.1', 'HTTP/2')"
+
+
+def test_invalid_version():
+ with pytest.raises(ValueError):
+ httpx.Version("HTTP/3")
projects / thirdparty / httpx.git / commitdiff
