SNOW-2463378: no_proxy support

DESCRIPTION.md

@@ -12,6 +12,7 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
     - This allows headless environments (like Docker or Airflow) running locally to auth via a browser URL.
   - Fix compilation error when building from sources with libc++.
   - Pin lower versions of dependencies to oldest version without vulnerabilities.
+  - Added no_proxy parameter for proxy configuration without using environmental variables.
 
 - v4.0.0(October 09,2025)
   - Added support for checking certificates revocation using revocation lists (CRLs)

src/snowflake/connector/auth/_auth.py

@@ -55,6 +55,7 @@
 from ..platform_detection import detect_platforms
 from ..session_manager import BaseHttpConfig, HttpConfig
 from ..session_manager import SessionManager as SyncSessionManager
+from ..session_manager import SessionManagerFactory
 from ..sqlstate import SQLSTATE_CONNECTION_WAS_NOT_ESTABLISHED
 from ..token_cache import TokenCache, TokenKey, TokenType
 from ..version import VERSION
@@ -116,7 +117,7 @@ def base_auth_data(
             # Extract base fields (automatically excludes subclass-specific fields)
             # Note: It won't be possible to pass adapter_factory from outer async-code to this part of code
             sync_config = HttpConfig(**http_config.to_base_dict())
-            session_manager = SyncSessionManager(config=sync_config)
+            session_manager = SessionManagerFactory.get_manager(config=sync_config)
 
         return {
             "data": {

src/snowflake/connector/connection.py

@@ -129,7 +129,12 @@
     ReauthenticationRequest,
     SnowflakeRestful,
 )
-from .session_manager import HttpConfig, ProxySupportAdapterFactory, SessionManager
+from .session_manager import (
+    HttpConfig,
+    ProxySupportAdapterFactory,
+    SessionManager,
+    SessionManagerFactory,
+)
 from .sqlstate import SQLSTATE_CONNECTION_NOT_EXISTS, SQLSTATE_FEATURE_NOT_SUPPORTED
 from .telemetry import TelemetryClient, TelemetryData, TelemetryField
 from .time_util import HeartBeatTimer, get_time_millis
@@ -199,6 +204,10 @@ def _get_private_bytes_from_file(
     "proxy_port": (None, (type(None), str)),  # snowflake
     "proxy_user": (None, (type(None), str)),  # snowflake
     "proxy_password": (None, (type(None), str)),  # snowflake
+    "no_proxy": (
+        None,
+        (type(None), str, Iterable),
+    ),  # hosts/ips to bypass proxy (str or iterable)
     "protocol": ("https", str),  # snowflake
     "warehouse": (None, (type(None), str)),  # snowflake
     "region": (None, (type(None), str)),  # snowflake
@@ -808,6 +817,10 @@ def proxy_user(self) -> str | None:
     def proxy_password(self) -> str | None:
         return self._proxy_password
 
+    @property
+    def no_proxy(self) -> str | Iterable | None:
+        return self._no_proxy
+
     @property
     def account(self) -> str:
         return self._account
@@ -1076,8 +1089,17 @@ def connect(self, **kwargs) -> None:
             proxy_port=self.proxy_port,
             proxy_user=self.proxy_user,
             proxy_password=self.proxy_password,
+            no_proxy=(
+                ",".join(str(x) for x in self.no_proxy)
+                if (
+                    self.no_proxy is not None
+                    and isinstance(self.no_proxy, Iterable)
+                    and not isinstance(self.no_proxy, (str, bytes))
+                )
+                else self.no_proxy
+            ),
         )
-        self._session_manager = SessionManager(self._http_config)
+        self._session_manager = SessionManagerFactory.get_manager(self._http_config)
 
         if self.enable_connection_diag:
             exceptions_dict = {}

src/snowflake/connector/connection_diagnostic.py

@@ -19,7 +19,7 @@
 
 from .compat import IS_WINDOWS, urlparse
 from .cursor import SnowflakeCursor
-from .session_manager import SessionManager
+from .session_manager import SessionManager, SessionManagerFactory
 from .url_util import extract_top_level_domain_from_hostname
 from .vendored import urllib3
 
@@ -197,7 +197,7 @@ def __init__(
         self._session_manager = (
             session_manager.clone(use_pooling=False)
             if session_manager
-            else SessionManager(use_pooling=False)
+            else SessionManagerFactory.get_manager(use_pooling=False)
         )
 
     def __parse_proxy(self, proxy_url: str) -> tuple[str, str, str, str]:

src/snowflake/connector/network.py

@@ -81,7 +81,12 @@
     ServiceUnavailableError,
     TooManyRequests,
 )
-from .session_manager import ProxySupportAdapterFactory, SessionManager, SessionPool
+from .session_manager import (
+    ProxySupportAdapterFactory,
+    SessionManager,
+    SessionManagerFactory,
+    SessionPool,
+)
 from .sqlstate import (
     SQLSTATE_CONNECTION_NOT_EXISTS,
     SQLSTATE_CONNECTION_REJECTED,
@@ -324,7 +329,9 @@ def __init__(
             session_manager = (
                 connection._session_manager
                 if (connection and connection._session_manager)
-                else SessionManager(adapter_factory=ProxySupportAdapterFactory())
+                else SessionManagerFactory.get_manager(
+                    adapter_factory=ProxySupportAdapterFactory()
+                )
             )
         self._session_manager = session_manager
         self._lock_token = Lock()
@@ -1213,5 +1220,5 @@ def _request_exec(
         except Exception as err:
             raise err
 
-    def use_session(self, url=None) -> Generator[Session, Any, None]:
+    def use_session(self, url: str | bytes) -> Generator[Session, Any, None]:
         return self.session_manager.use_session(url)

src/snowflake/connector/ocsp_snowflake.py

@@ -59,6 +59,7 @@
 from .backoff_policies import exponential_backoff
 from .cache import CacheEntry, SFDictCache, SFDictFileCache
 from .constants import OCSP_ROOT_CERTS_DICT_LOCK_TIMEOUT_DEFAULT_NO_TIMEOUT
+from .session_manager import SessionManagerFactory
 from .telemetry import TelemetryField, generate_telemetry_data_dict
 from .url_util import extract_top_level_domain_from_hostname, url_encode_str
 from .util_text import _base64_bytes_to_str
@@ -551,8 +552,8 @@ def _download_ocsp_response_cache(ocsp, url, do_retry: bool = True) -> bool:
             # Obtain SessionManager from ssl_wrap_socket context var if available
             session_manager = get_current_session_manager(
                 use_pooling=False
-            ) or SessionManager(use_pooling=False)
-            with session_manager.use_session() as session:
+            ) or SessionManagerFactory.get_manager(use_pooling=False)
+            with session_manager.use_session(url) as session:
                 max_retry = SnowflakeOCSP.OCSP_CACHE_SERVER_MAX_RETRY if do_retry else 1
                 sleep_time = 1
                 backoff = exponential_backoff()()
@@ -1646,9 +1647,9 @@ def _fetch_ocsp_response(
         session_manager: SessionManager = (
             context_session_manager
             if context_session_manager is not None
-            else SessionManager(use_pooling=False)
+            else SessionManagerFactory.get_manager(use_pooling=False)
         )
-        with session_manager.use_session() as session:
+        with session_manager.use_session(target_url) as session:
             max_retry = sf_max_retry if do_retry else 1
             sleep_time = 1
             backoff = exponential_backoff()()

src/snowflake/connector/platform_detection.py

@@ -13,7 +13,7 @@
     Config = botocore.config.Config
     IMDSFetcher = botocore.utils.IMDSFetcher
 
-from .session_manager import SessionManager
+from .session_manager import SessionManager, SessionManagerFactory
 from .vendored.requests import RequestException, Timeout
 
 logger = logging.getLogger(__name__)
@@ -415,7 +415,9 @@ def detect_platforms(
             logger.debug(
                 "No session manager provided. HTTP settings may not be preserved. Using default."
             )
-            session_manager = SessionManager(use_pooling=False, max_retries=0)
+            session_manager = SessionManagerFactory.get_manager(
+                use_pooling=False, max_retries=0
+            )
 
         # Run environment-only checks synchronously (no network calls, no threading overhead)
         platforms = {

src/snowflake/connector/result_batch.py

@@ -26,7 +26,7 @@
 from .options import installed_pandas
 from .options import pyarrow as pa
 from .secret_detector import SecretDetector
-from .session_manager import HttpConfig, SessionManager
+from .session_manager import HttpConfig, SessionManager, SessionManagerFactory
 from .time_util import TimerContextManager
 
 logger = getLogger(__name__)
@@ -319,7 +319,7 @@ def http_config(self, config: HttpConfig) -> None:
         if self._session_manager:
             self._session_manager.config = config
         else:
-            self._session_manager = SessionManager(config=config)
+            self._session_manager = SessionManagerFactory.get_manager(config=config)
 
     def __iter__(
         self,
@@ -360,21 +360,27 @@ def _download(
                         and connection.rest.session_manager is not None
                     ):
                         # If connection was explicitly passed and not closed yet - we can reuse SessionManager with session pooling
-                        with connection.rest.use_session() as session:
+                        with connection.rest.use_session(
+                            request_data["url"]
+                        ) as session:
                             logger.debug(
                                 f"downloading result batch id: {self.id} with existing session {session}"
                             )
                             response = session.request("get", **request_data)
                     elif self._session_manager is not None:
                         # If connection is not accessible or was already closed, but cursors are now used to fetch the data - we will only reuse the http setup (through cloned SessionManager without session pooling)
-                        with self._session_manager.use_session() as session:
+                        with self._session_manager.use_session(
+                            request_data["url"]
+                        ) as session:
                             response = session.request("get", **request_data)
                     else:
                         # If there was no session manager cloned, then we are using a default Session Manager setup, since it is very unlikely to enter this part outside of testing
                         logger.debug(
                             f"downloading result batch id: {self.id} with new session through local session manager"
                         )
-                        local_session_manager = SessionManager(use_pooling=False)
+                        local_session_manager = SessionManagerFactory.get_manager(
+                            use_pooling=False
+                        )
                         response = local_session_manager.get(**request_data)
 
                     if response.status_code == OK:

src/snowflake/connector/session_manager.py

@@ -134,6 +134,7 @@ class BaseHttpConfig:
     proxy_port: str | None = None
     proxy_user: str | None = None
     proxy_password: str | None = None
+    no_proxy: str | None = None
 
     def copy_with(self, **overrides: Any) -> BaseHttpConfig:
         """Return a new config with overrides applied."""
@@ -191,12 +192,12 @@ def __init__(self, manager: SessionManager) -> None:
         self._active_sessions: set[SessionT] = set()
         self._manager = manager
 
-    def get_session(self) -> SessionT:
+    def get_session(self, *, url: str | None = None) -> SessionT:
         """Returns a session from the session pool or creates a new one."""
         try:
             session = self._idle_sessions.pop()
         except IndexError:
-            session = self._manager.make_session()
+            session = self._manager.make_session(url=url)
         self._active_sessions.add(session)
         return session
 
@@ -403,6 +404,7 @@ def __init__(self, config: HttpConfig | None = None, **http_config_kwargs) -> No
             logger.debug("Creating a config for the SessionManager")
             config = HttpConfig(**http_config_kwargs)
         self._cfg: HttpConfig = config
+        # Maps hostname to SessionPool instance for its connections
         self._sessions_map: dict[str | None, SessionPool] = collections.defaultdict(
             lambda: SessionPool(self)
         )
@@ -474,32 +476,50 @@ def _mount_adapters(self, session: requests.Session) -> None:
             )
             return
 
-    def make_session(self) -> Session:
+    def make_session(self, *, url: str | None = None) -> Session:
         session = requests.Session()
         self._mount_adapters(session)
-        session.proxies = {"http": self.proxy_url, "https": self.proxy_url}
         return session
 
     @contextlib.contextmanager
     @_propagate_session_manager_to_ocsp
     def use_session(
-        self, url: str | bytes | None = None, use_pooling: bool | None = None
+        self, url: str | bytes, use_pooling: bool | None = None
     ) -> Generator[Session, Any, None]:
+        """
+        'url' is an obligatory parameter due to the need for correct proxy handling (i.e. bypassing caused by no_proxy settings).
+        """
         use_pooling = use_pooling if use_pooling is not None else self.use_pooling
         if not use_pooling:
-            session = self.make_session()
+            session = self.make_session(url=url)
             try:
                 yield session
             finally:
                 session.close()
         else:
-            hostname = urlparse(url).hostname if url else None
-            pool = self._sessions_map[hostname]
-            session = pool.get_session()
-            try:
-                yield session
-            finally:
-                pool.return_session(session)
+            yield from self._yield_session_from_pool(url)
+
+    def _yield_session_from_pool(
+        self, url: str | bytes
+    ) -> Generator[SessionT, Any, None]:
+        hostname = self._get_pooling_key_from_url(url)
+        pool = self._sessions_map[hostname]
+        session = pool.get_session(url=url)
+        try:
+            yield session
+        finally:
+            pool.return_session(session)
+
+    @staticmethod
+    def _get_pooling_key_from_url(url: str) -> str | None:
+        """
+        Derive the session pooling key (hostname) from a URL.
+
+        :param url: Absolute URL the session will be used for.
+        :return: Hostname string or None if URL is missing/invalid.
+        """
+        hostname = urlparse(url).hostname if url else None
+        return hostname
 
     def request(
         self,
@@ -586,3 +606,42 @@ def request(
         use_pooling=use_pooling,
         **kwargs,
     )
+
+
+class ProxySessionManager(SessionManager):
+    def make_session(self, *, url: str | None = None) -> Session:
+        session = requests.Session()
+        self._mount_adapters(session)
+        proxies = (
+            {
+                "no_proxy": self._cfg.no_proxy,
+            }
+            if requests.utils.should_bypass_proxies(url, no_proxy=self.config.no_proxy)
+            else {
+                "http": self.proxy_url,
+                "https": self.proxy_url,
+                "no_proxy": self.config.no_proxy,
+            }
+        )
+        session.proxies = proxies
+        return session
+
+    def clone(
+        self,
+        **http_config_overrides,
+    ) -> SessionManager:
+        return ProxySessionManager.from_config(self._cfg, **http_config_overrides)
+
+
+class SessionManagerFactory:
+    @staticmethod
+    def get_manager(
+        config: HttpConfig | None = None, **http_config_kwargs
+    ) -> SessionManager:
+        has_param_proxies = (
+            hasattr(config, "proxy_host") or "proxies" in http_config_kwargs
+        )
+        if has_param_proxies:
+            return ProxySessionManager(config, **http_config_kwargs)
+        else:
+            return SessionManager(config, **http_config_kwargs)

src/snowflake/connector/ssl_wrap_socket.py

@@ -25,7 +25,7 @@
 from .crl import CertRevocationCheckMode, CRLConfig, CRLValidator
 from .errorcode import ER_OCSP_RESPONSE_CERT_STATUS_REVOKED
 from .errors import OperationalError
-from .session_manager import SessionManager
+from .session_manager import SessionManager, SessionManagerFactory
 from .vendored.urllib3 import connection as connection_
 from .vendored.urllib3.contrib.pyopenssl import PyOpenSSLContext, WrappedSocket
 from .vendored.urllib3.util import ssl_ as ssl_
@@ -115,11 +115,15 @@ def get_current_session_manager(
     """
     sm_weak_ref = _CURRENT_SESSION_MANAGER.get()
     if sm_weak_ref is None:
-        return SessionManager() if create_default_if_missing else None
+        return (
+            SessionManagerFactory.get_manager() if create_default_if_missing else None
+        )
     context_session_manager = sm_weak_ref()
 
     if context_session_manager is None:
-        return SessionManager() if create_default_if_missing else None
+        return (
+            SessionManagerFactory.get_manager() if create_default_if_missing else None
+        )
 
     return context_session_manager.clone(**clone_kwargs)