Add support for the no_proxy env var mechanism in the HTTP client

CHANGES/4431.bugfix

@@ -0,0 +1 @@
+Fixed HTTP client requests to honor ``no_proxy`` environment variables.

CONTRIBUTORS.txt

@@ -265,6 +265,7 @@ Stanislav Prokop
 Stefan Tjarks
 Stepan Pletnev
 Stephan Jaensch
+Stephen Cirelli
 Stephen Granade
 Steven Seguin
 Sunghyun Hwang

aiohttp/client.py

@@ -9,6 +9,7 @@
 import sys
 import traceback
 import warnings
+from contextlib import suppress
 from types import SimpleNamespace, TracebackType
 from typing import (
     Any,
@@ -80,7 +81,7 @@
     BasicAuth,
     TimeoutHandle,
     ceil_timeout,
-    proxies_from_env,
+    get_env_proxy_for_url,
     sentinel,
     strip_auth_from_url,
 )
@@ -444,11 +445,8 @@ async def _request(
                     if proxy is not None:
                         proxy = URL(proxy)
                     elif self._trust_env:
-                        for scheme, proxy_info in proxies_from_env().items():
-                            if scheme == url.scheme:
-                                proxy = proxy_info.proxy
-                                proxy_auth = proxy_info.proxy_auth
-                                break
+                        with suppress(LookupError):
+                            proxy, proxy_auth = get_env_proxy_for_url(url)
 
                     req = self._request_class(
                         method,

aiohttp/helpers.py

@@ -40,7 +40,7 @@
     cast,
 )
 from urllib.parse import quote
-from urllib.request import getproxies
+from urllib.request import getproxies, proxy_bypass
 
 import async_timeout
 from multidict import CIMultiDict, MultiDict, MultiDictProxy
@@ -269,6 +269,20 @@ def proxies_from_env() -> Dict[str, ProxyInfo]:
     return ret
 
 
+def get_env_proxy_for_url(url: URL) -> Tuple[URL, Optional[BasicAuth]]:
+    """Get a permitted proxy for the given URL from the env."""
+    if url.host is not None and proxy_bypass(url.host):
+        raise LookupError(f"Proxying is disallowed for `{url.host!r}`")
+
+    proxies_in_env = proxies_from_env()
+    try:
+        proxy_info = proxies_in_env[url.scheme]
+    except KeyError:
+        raise LookupError(f"No proxies found for `{url!s}` in the env")
+    else:
+        return proxy_info.proxy, proxy_info.proxy_auth
+
+
 @dataclasses.dataclass(frozen=True)
 class MimeType:
     type: str

tests/test_helpers.py

@@ -515,6 +515,77 @@ def test_proxies_from_env_http_with_auth(mocker) -> None:
     assert proxy_auth.encoding == "latin1"
 
 
+# --------------------- get_env_proxy_for_url ------------------------------
+
+
+@pytest.mark.parametrize(
+    ("bypass", "proxies", "url_input", "expected_err_msg"),
+    (
+        (
+            True,
+            {"http": helpers.ProxyInfo("http://example.com", "")},
+            "http://aiohttp.io/path",
+            r"Proxying is disallowed for `'aiohttp.io'`",
+        ),
+        (
+            False,
+            {"https": helpers.ProxyInfo("https://example.com", "")},
+            "http://aiohttp.io/path",
+            r"No proxies found for `http://aiohttp.io/path` in the env",
+        ),
+        (
+            False,
+            {"https": helpers.ProxyInfo("https://example.com", "")},
+            "",
+            r"No proxies found for `` in the env",
+        ),
+    ),
+    ids=(
+        "url_matches_the_no_proxy_list",
+        "url_scheme_does_not_match_proxy_list",
+        "url_is_empty",
+    ),
+)
+def test_get_env_proxy_for_url_negative(
+    bypass,
+    expected_err_msg,
+    mocker,
+    proxies,
+    url_input,
+) -> None:
+    mocker.patch("aiohttp.helpers.proxy_bypass", return_value=bypass)
+    mocker.patch("aiohttp.helpers.proxies_from_env", return_value=proxies)
+    url = URL(url_input)
+    with pytest.raises(LookupError, match=expected_err_msg):
+        helpers.get_env_proxy_for_url(url)
+
+
+@pytest.mark.parametrize(
+    ("schema", "proxyInfo", "url"),
+    [
+        (
+            "https",
+            helpers.ProxyInfo("https://other.com", ""),
+            URL("https://aiohttp.com"),
+        ),
+        (
+            "http",
+            helpers.ProxyInfo("http://example.com", ""),
+            URL("http://aiohttp.com"),
+        ),
+    ],
+    ids=("https", "http"),
+)
+def test_get_env_proxy_for_url_when_url_scheme_matches_proxy_list(
+    mocker, schema, proxyInfo, url
+) -> None:
+    mocker.patch("aiohttp.helpers.proxy_bypass", return_value=False)
+    mocker.patch("aiohttp.helpers.proxies_from_env", return_value={schema: proxyInfo})
+    proxy, proxy_auth = helpers.get_env_proxy_for_url(url)
+    assert proxy == proxyInfo.proxy
+    assert proxy_auth == proxyInfo.proxy_auth
+
+
 # ------------- set_result / set_exception ----------------------