[python] honor proxy environment settings

modules/openapi-generator/src/main/resources/python/configuration.mustache

@@ -16,6 +16,10 @@ import multiprocessing
 {{/async}}
 import sys
 from typing import Any, ClassVar, Dict, List, Literal, Optional, TypedDict, Union
+{{^async}}
+from urllib.parse import urlparse
+from urllib.request import getproxies
+{{/async}}
 from typing_extensions import NotRequired, Self
 
 {{^async}}
@@ -215,6 +219,9 @@ class Configuration:
     :param tls_server_name: SSL/TLS Server Name Indication (SNI). Set this to the SNI value expected by the server.
     :param connection_pool_maxsize: Connection pool max size. None in the constructor is coerced to 100 for async and cpu_count * 5 for sync.
     :param proxy: Proxy URL.
+{{^async}}
+    :param no_proxy: Comma-separated hosts that bypass the proxy.
+{{/async}}
     :param proxy_headers: Proxy headers.
     :param safe_chars_for_path_param: Safe characters for path parameter encoding.
     :param client_side_validation: Enable client-side validation. Default True.
@@ -346,6 +353,9 @@ conf = {{{packageName}}}.Configuration(
         tls_server_name: Optional[str]=None,
         connection_pool_maxsize: Optional[int]=None,
         proxy: Optional[str]=None,
+{{^async}}
+        no_proxy: Optional[str]=None,
+{{/async}}
         proxy_headers: Optional[Any]=None,
         safe_chars_for_path_param: str='',
         client_side_validation: bool=True,
@@ -469,9 +479,25 @@ conf = {{{packageName}}}.Configuration(
         """
         {{/async}}
 
+{{^async}}
+        # urllib3 does not read proxy environment variables itself:
+        # https://github.com/urllib3/urllib3/issues/1785
+        if proxy is None or no_proxy is None:
+            proxies = getproxies()
+            if proxy is None:
+                scheme = urlparse(self.host).scheme
+                proxy = proxies.get(scheme) or proxies.get("all")
+            if no_proxy is None:
+                no_proxy = proxies.get("no")
+{{/async}}
         self.proxy = proxy
         """Proxy URL
         """
+{{^async}}
+        self.no_proxy = no_proxy
+        """Hosts that bypass the proxy
+        """
+{{/async}}
         self.proxy_headers = proxy_headers
         """Proxy headers
         """

modules/openapi-generator/src/main/resources/python/rest.mustache

@@ -3,10 +3,12 @@
 {{>partial_header}}
 
 
+import ipaddress
 import io
 import json
 import re
 import ssl
+from urllib.parse import urlparse
 
 import urllib3
 
@@ -26,6 +28,44 @@ def is_socks_proxy_url(url):
         return split_section[0].lower() in SUPPORTED_SOCKS_PROXIES
 
 
+def should_bypass_proxies(url: str, no_proxy: str) -> bool:
+    """Return whether ``url`` matches the comma-separated ``no_proxy`` rules."""
+    parsed_url = urlparse(url)
+    if not parsed_url.hostname:
+        return True
+
+    host = parsed_url.hostname.lower()
+    host_and_port = parsed_url.netloc.lower()
+    try:
+        host_ip = ipaddress.ip_address(host)
+    except ValueError:
+        host_ip = None
+
+    for entry in (entry.strip().lower() for entry in no_proxy.split(',')):
+        if not entry:
+            continue
+        if entry == '*':
+            return True
+
+        if host_ip is not None:
+            try:
+                if host_ip in ipaddress.ip_network(entry, strict=False):
+                    return True
+            except ValueError:
+                pass
+
+        entry = entry.lstrip('.')
+        if (
+            host == entry
+            or host.endswith('.' + entry)
+            or host_and_port == entry
+            or host_and_port.endswith('.' + entry)
+        ):
+            return True
+
+    return False
+
+
 class RESTResponse(io.IOBase):
 
     def __init__(self, resp) -> None:
@@ -95,7 +135,9 @@ class RESTClientObject:
         # https pool manager
         self.pool_manager: urllib3.PoolManager
 
-        if configuration.proxy:
+        if configuration.proxy and not should_bypass_proxies(
+            configuration.host, configuration.no_proxy or ''
+        ):
             if is_socks_proxy_url(configuration.proxy):
                 from urllib3.contrib.socks import SOCKSProxyManager
                 pool_args["proxy_url"] = configuration.proxy

samples/client/echo_api/python-disallowAdditionalPropertiesIfNotPresent/openapi_client/configuration.py

@@ -18,6 +18,8 @@
 import multiprocessing
 import sys
 from typing import Any, ClassVar, Dict, List, Literal, Optional, TypedDict, Union
+from urllib.parse import urlparse
+from urllib.request import getproxies
 from typing_extensions import NotRequired, Self
 
 import urllib3
@@ -171,6 +173,7 @@ class Configuration:
     :param tls_server_name: SSL/TLS Server Name Indication (SNI). Set this to the SNI value expected by the server.
     :param connection_pool_maxsize: Connection pool max size. None in the constructor is coerced to 100 for async and cpu_count * 5 for sync.
     :param proxy: Proxy URL.
+    :param no_proxy: Comma-separated hosts that bypass the proxy.
     :param proxy_headers: Proxy headers.
     :param safe_chars_for_path_param: Safe characters for path parameter encoding.
     :param client_side_validation: Enable client-side validation. Default True.
@@ -222,6 +225,7 @@ def __init__(
         tls_server_name: Optional[str]=None,
         connection_pool_maxsize: Optional[int]=None,
         proxy: Optional[str]=None,
+        no_proxy: Optional[str]=None,
         proxy_headers: Optional[Any]=None,
         safe_chars_for_path_param: str='',
         client_side_validation: bool=True,
@@ -328,9 +332,21 @@ def __init__(
            per pool. None in the constructor is coerced to cpu_count * 5.
         """
 
+        # urllib3 does not read proxy environment variables itself:
+        # https://github.com/urllib3/urllib3/issues/1785
+        if proxy is None or no_proxy is None:
+            proxies = getproxies()
+            if proxy is None:
+                scheme = urlparse(self.host).scheme
+                proxy = proxies.get(scheme) or proxies.get("all")
+            if no_proxy is None:
+                no_proxy = proxies.get("no")
         self.proxy = proxy
         """Proxy URL
         """
+        self.no_proxy = no_proxy
+        """Hosts that bypass the proxy
+        """
         self.proxy_headers = proxy_headers
         """Proxy headers
         """

samples/client/echo_api/python-disallowAdditionalPropertiesIfNotPresent/openapi_client/rest.py

@@ -13,10 +13,12 @@
 """  # noqa: E501
 
 
+import ipaddress
 import io
 import json
 import re
 import ssl
+from urllib.parse import urlparse
 
 import urllib3
 
@@ -36,6 +38,44 @@ def is_socks_proxy_url(url):
         return split_section[0].lower() in SUPPORTED_SOCKS_PROXIES
 
 
+def should_bypass_proxies(url: str, no_proxy: str) -> bool:
+    """Return whether ``url`` matches the comma-separated ``no_proxy`` rules."""
+    parsed_url = urlparse(url)
+    if not parsed_url.hostname:
+        return True
+
+    host = parsed_url.hostname.lower()
+    host_and_port = parsed_url.netloc.lower()
+    try:
+        host_ip = ipaddress.ip_address(host)
+    except ValueError:
+        host_ip = None
+
+    for entry in (entry.strip().lower() for entry in no_proxy.split(',')):
+        if not entry:
+            continue
+        if entry == '*':
+            return True
+
+        if host_ip is not None:
+            try:
+                if host_ip in ipaddress.ip_network(entry, strict=False):
+                    return True
+            except ValueError:
+                pass
+
+        entry = entry.lstrip('.')
+        if (
+            host == entry
+            or host.endswith('.' + entry)
+            or host_and_port == entry
+            or host_and_port.endswith('.' + entry)
+        ):
+            return True
+
+    return False
+
+
 class RESTResponse(io.IOBase):
 
     def __init__(self, resp) -> None:
@@ -105,7 +145,9 @@ def __init__(self, configuration) -> None:
         # https pool manager
         self.pool_manager: urllib3.PoolManager
 
-        if configuration.proxy:
+        if configuration.proxy and not should_bypass_proxies(
+            configuration.host, configuration.no_proxy or ''
+        ):
             if is_socks_proxy_url(configuration.proxy):
                 from urllib3.contrib.socks import SOCKSProxyManager
                 pool_args["proxy_url"] = configuration.proxy

samples/client/echo_api/python/openapi_client/configuration.py

@@ -18,6 +18,8 @@
 import multiprocessing
 import sys
 from typing import Any, ClassVar, Dict, List, Literal, Optional, TypedDict, Union
+from urllib.parse import urlparse
+from urllib.request import getproxies
 from typing_extensions import NotRequired, Self
 
 import urllib3
@@ -171,6 +173,7 @@ class Configuration:
     :param tls_server_name: SSL/TLS Server Name Indication (SNI). Set this to the SNI value expected by the server.
     :param connection_pool_maxsize: Connection pool max size. None in the constructor is coerced to 100 for async and cpu_count * 5 for sync.
     :param proxy: Proxy URL.
+    :param no_proxy: Comma-separated hosts that bypass the proxy.
     :param proxy_headers: Proxy headers.
     :param safe_chars_for_path_param: Safe characters for path parameter encoding.
     :param client_side_validation: Enable client-side validation. Default True.
@@ -222,6 +225,7 @@ def __init__(
         tls_server_name: Optional[str]=None,
         connection_pool_maxsize: Optional[int]=None,
         proxy: Optional[str]=None,
+        no_proxy: Optional[str]=None,
         proxy_headers: Optional[Any]=None,
         safe_chars_for_path_param: str='',
         client_side_validation: bool=True,
@@ -328,9 +332,21 @@ def __init__(
            per pool. None in the constructor is coerced to cpu_count * 5.
         """
 
+        # urllib3 does not read proxy environment variables itself:
+        # https://github.com/urllib3/urllib3/issues/1785
+        if proxy is None or no_proxy is None:
+            proxies = getproxies()
+            if proxy is None:
+                scheme = urlparse(self.host).scheme
+                proxy = proxies.get(scheme) or proxies.get("all")
+            if no_proxy is None:
+                no_proxy = proxies.get("no")
         self.proxy = proxy
         """Proxy URL
         """
+        self.no_proxy = no_proxy
+        """Hosts that bypass the proxy
+        """
         self.proxy_headers = proxy_headers
         """Proxy headers
         """

samples/client/echo_api/python/openapi_client/rest.py

@@ -13,10 +13,12 @@
 """  # noqa: E501
 
 
+import ipaddress
 import io
 import json
 import re
 import ssl
+from urllib.parse import urlparse
 
 import urllib3
 
@@ -36,6 +38,44 @@ def is_socks_proxy_url(url):
         return split_section[0].lower() in SUPPORTED_SOCKS_PROXIES
 
 
+def should_bypass_proxies(url: str, no_proxy: str) -> bool:
+    """Return whether ``url`` matches the comma-separated ``no_proxy`` rules."""
+    parsed_url = urlparse(url)
+    if not parsed_url.hostname:
+        return True
+
+    host = parsed_url.hostname.lower()
+    host_and_port = parsed_url.netloc.lower()
+    try:
+        host_ip = ipaddress.ip_address(host)
+    except ValueError:
+        host_ip = None
+
+    for entry in (entry.strip().lower() for entry in no_proxy.split(',')):
+        if not entry:
+            continue
+        if entry == '*':
+            return True
+
+        if host_ip is not None:
+            try:
+                if host_ip in ipaddress.ip_network(entry, strict=False):
+                    return True
+            except ValueError:
+                pass
+
+        entry = entry.lstrip('.')
+        if (
+            host == entry
+            or host.endswith('.' + entry)
+            or host_and_port == entry
+            or host_and_port.endswith('.' + entry)
+        ):
+            return True
+
+    return False
+
+
 class RESTResponse(io.IOBase):
 
     def __init__(self, resp) -> None:
@@ -105,7 +145,9 @@ def __init__(self, configuration) -> None:
         # https pool manager
         self.pool_manager: urllib3.PoolManager
 
-        if configuration.proxy:
+        if configuration.proxy and not should_bypass_proxies(
+            configuration.host, configuration.no_proxy or ''
+        ):
             if is_socks_proxy_url(configuration.proxy):
                 from urllib3.contrib.socks import SOCKSProxyManager
                 pool_args["proxy_url"] = configuration.proxy