* Drop ASGIDispatch, WSGIDispatch, which has been replaced by ASGITransport, WSGITransport.
* Drop dispatch=... on client, which has been replaced by transport=...
* Drop soft_limit, hard_limit, which have been replaced by max_keepalive and max_connections.
* Drop Response.stream and Response.raw, which have been replaced by aiter_bytes and aiter_raw.
* Drop internal usage of as_network_error in favour of map_exceptions.
)
from ._models import URL, Cookies, Headers, QueryParams, Request, Response
from ._status_codes import StatusCode, codes
-from ._transports.asgi import ASGIDispatch, ASGITransport
+from ._transports.asgi import ASGITransport
from ._transports.urllib3 import URLLib3ProxyTransport, URLLib3Transport
-from ._transports.wsgi import WSGIDispatch, WSGITransport
+from ._transports.wsgi import WSGITransport
__all__ = [
"__description__",
"request",
"stream",
"codes",
- "ASGIDispatch",
"ASGITransport",
"AsyncClient",
"Auth",
"Request",
"Response",
"DigestAuth",
- "WSGIDispatch",
"WSGITransport",
]
get_environment_proxies,
get_logger,
should_not_be_proxied,
- warn_deprecated,
)
logger = get_logger(__name__)
request URLs.
* **transport** - *(optional)* A transport class to use for sending requests
over the network.
- * **dispatch** - *(optional)* A deprecated alias for transport.
* **app** - *(optional)* An WSGI application to send requests to,
rather than sending actual network requests.
* **trust_env** - *(optional)* Enables or disables usage of environment
max_redirects: int = DEFAULT_MAX_REDIRECTS,
base_url: URLTypes = None,
transport: httpcore.SyncHTTPTransport = None,
- dispatch: httpcore.SyncHTTPTransport = None,
app: typing.Callable = None,
trust_env: bool = True,
):
proxy_map = self.get_proxy_map(proxies, trust_env)
- if dispatch is not None:
- warn_deprecated(
- "The dispatch argument is deprecated since v0.13 and will be "
- "removed in a future release, please use 'transport'"
- )
- if transport is None:
- transport = dispatch
-
self.transport = self.init_transport(
verify=verify,
cert=cert,
request URLs.
* **transport** - *(optional)* A transport class to use for sending requests
over the network.
- * **dispatch** - *(optional)* A deprecated alias for transport.
* **app** - *(optional)* An ASGI application to send requests to,
rather than sending actual network requests.
* **trust_env** - *(optional)* Enables or disables usage of environment
max_redirects: int = DEFAULT_MAX_REDIRECTS,
base_url: URLTypes = None,
transport: httpcore.AsyncHTTPTransport = None,
- dispatch: httpcore.AsyncHTTPTransport = None,
app: typing.Callable = None,
trust_env: bool = True,
):
trust_env=trust_env,
)
- if dispatch is not None:
- warn_deprecated(
- "The dispatch argument is deprecated since v0.13 and will be "
- "removed in a future release, please use 'transport'",
- )
- if transport is None:
- transport = dispatch
-
proxy_map = self.get_proxy_map(proxies, trust_env)
self.transport = self.init_transport(
from ._models import URL, Headers
from ._types import CertTypes, HeaderTypes, TimeoutTypes, URLTypes, VerifyTypes
-from ._utils import get_ca_bundle_from_env, get_logger, warn_deprecated
+from ._utils import get_ca_bundle_from_env, get_logger
DEFAULT_CIPHERS = ":".join(
[
"""
def __init__(
- self,
- *,
- max_keepalive: int = None,
- max_connections: int = None,
- soft_limit: int = None,
- hard_limit: int = None,
+ self, *, max_keepalive: int = None, max_connections: int = None,
):
self.max_keepalive = max_keepalive
self.max_connections = max_connections
- if soft_limit is not None: # pragma: nocover
- self.max_keepalive = soft_limit
- warn_deprecated("'soft_limit' is deprecated. Use 'max_keepalive' instead.",)
- if hard_limit is not None: # pragma: nocover
- self.max_connections = hard_limit
- warn_deprecated(
- "'hard_limit' is deprecated. Use 'max_connections' instead.",
- )
def __eq__(self, other: typing.Any) -> bool:
return (
obfuscate_sensitive_headers,
parse_header_links,
str_query_param,
- warn_deprecated,
)
def __repr__(self) -> str:
return f"<Response [{self.status_code} {self.reason_phrase}]>"
- @property
- def stream(self): # type: ignore
- warn_deprecated( # pragma: nocover
- "Response.stream() is due to be deprecated. "
- "Use Response.aiter_bytes() instead.",
- )
- return self.aiter_bytes # pragma: nocover
-
- @property
- def raw(self): # type: ignore
- warn_deprecated( # pragma: nocover
- "Response.raw() is due to be deprecated. "
- "Use Response.aiter_raw() instead.",
- )
- return self.aiter_raw # pragma: nocover
-
def read(self) -> bytes:
"""
Read and return the response content.
import sniffio
from .._content_streams import ByteStream
-from .._utils import warn_deprecated
if TYPE_CHECKING: # pragma: no cover
import asyncio
stream = ByteStream(b"".join(body_parts))
return (b"HTTP/1.1", status_code, b"", response_headers, stream)
-
-
-class ASGIDispatch(ASGITransport):
- def __init__(
- self,
- app: Callable,
- raise_app_exceptions: bool = True,
- root_path: str = "",
- client: Tuple[str, int] = ("127.0.0.1", 123),
- ) -> None:
- warn_deprecated("ASGIDispatch is deprecated, please use ASGITransport")
- super().__init__(
- app=app,
- raise_app_exceptions=raise_app_exceptions,
- root_path=root_path,
- client=client,
- )
from .._config import Proxy, SSLConfig
from .._content_streams import ByteStream, IteratorStream
+from .._exceptions import NetworkError, map_exceptions
from .._types import CertTypes, VerifyTypes
-from .._utils import as_network_error
try:
import urllib3
path.decode("ascii"),
)
- with as_network_error(MaxRetryError, SSLError, socket.error):
+ with map_exceptions(
+ {
+ MaxRetryError: NetworkError,
+ SSLError: NetworkError,
+ socket.error: NetworkError,
+ }
+ ):
conn = self.pool.urlopen(
method=method.decode(),
url=url_str,
)
def response_bytes() -> Iterator[bytes]:
- with as_network_error(socket.error):
+ with map_exceptions({socket.error: NetworkError}):
for chunk in conn.stream(4096, decode_content=False):
yield chunk
import httpcore
from .._content_streams import ByteStream, IteratorStream
-from .._utils import warn_deprecated
def _skip_leading_empty_chunks(body: typing.Iterable) -> typing.Iterable:
stream = IteratorStream(chunk for chunk in result)
return (b"HTTP/1.1", status_code, b"", headers, stream)
-
-
-class WSGIDispatch(WSGITransport):
- def __init__(
- self,
- app: typing.Callable,
- raise_app_exceptions: bool = True,
- script_name: str = "",
- remote_addr: str = "127.0.0.1",
- ) -> None:
- warn_deprecated("WSGIDispatch is deprecated, please use WSGITransport")
- super().__init__(
- app=app,
- raise_app_exceptions=raise_app_exceptions,
- script_name=script_name,
- remote_addr=remote_addr,
- )
import codecs
import collections
-import contextlib
import logging
import mimetypes
import netrc
from types import TracebackType
from urllib.request import getproxies
-from ._exceptions import NetworkError
from ._types import PrimitiveData
if typing.TYPE_CHECKING: # pragma: no cover
return timedelta(seconds=self.end - self.start)
-@contextlib.contextmanager
-def as_network_error(*exception_classes: type) -> typing.Iterator[None]:
- try:
- yield
- except BaseException as exc:
- for cls in exception_classes:
- if isinstance(exc, cls):
- raise NetworkError(exc) from exc
- raise
-
-
-def warn_deprecated(message: str) -> None:
+def warn_deprecated(message: str) -> None: # pragma: nocover
warnings.warn(message, DeprecationWarning, stacklevel=2)
from datetime import timedelta
-import httpcore
import pytest
import httpx
-from httpx import ASGIDispatch
@pytest.mark.usefixtures("async_environment")
assert response.status_code == 200
assert response.content == data
-
-
-def test_dispatch_deprecated():
- dispatch = httpcore.AsyncHTTPTransport()
-
- with pytest.warns(DeprecationWarning) as record:
- client = httpx.AsyncClient(dispatch=dispatch)
-
- assert client.transport is dispatch
- assert len(record) == 1
- assert record[0].message.args[0] == (
- "The dispatch argument is deprecated since v0.13 and will be "
- "removed in a future release, please use 'transport'"
- )
-
-
-def test_asgi_dispatch_deprecated():
- async def app(scope, receive, send):
- pass
-
- with pytest.warns(DeprecationWarning) as record:
- ASGIDispatch(app)
-
- assert len(record) == 1
- assert (
- record[0].message.args[0]
- == "ASGIDispatch is deprecated, please use ASGITransport"
- )
from datetime import timedelta
-import httpcore
import pytest
import httpx
-from httpx import WSGIDispatch
def test_get(server):
assert url.scheme == "https"
assert url.is_ssl
-
-
-def test_dispatch_deprecated():
- dispatch = httpcore.SyncHTTPTransport()
-
- with pytest.warns(DeprecationWarning) as record:
- client = httpx.Client(dispatch=dispatch)
-
- assert client.transport is dispatch
- assert len(record) == 1
- assert record[0].message.args[0] == (
- "The dispatch argument is deprecated since v0.13 and will be "
- "removed in a future release, please use 'transport'"
- )
-
-
-def test_wsgi_dispatch_deprecated():
- def app(start_response, environ):
- pass
-
- with pytest.warns(DeprecationWarning) as record:
- WSGIDispatch(app)
-
- assert len(record) == 1
- assert (
- record[0].message.args[0]
- == "WSGIDispatch is deprecated, please use WSGITransport"
- )
and value not in HTTPCORE_EXC_MAP
]
- if not_mapped:
+ if not_mapped: # pragma: nocover
pytest.fail(f"Unmapped httpcore exceptions: {not_mapped}")
and not hasattr(httpx, name)
]
- if not_exposed:
+ if not_exposed: # pragma: nocover
pytest.fail(f"Unexposed HTTPX exceptions: {not_exposed}")
projects / thirdparty / httpx.git / commitdiff
