diff --git a/DESCRIPTION.md b/DESCRIPTION.md
index c49575c4d..03fe86320 100644
--- a/DESCRIPTION.md
+++ b/DESCRIPTION.md
@@ -7,6 +7,9 @@ https://docs.snowflake.com/
 Source code is also available at: https://github.com/snowflakedb/snowflake-connector-python
 
 # Release Notes
+- v4.2.1(TBD)
+  - Fixed a memory leak in `SnowflakeRestful.fetch()`.
+
 - v4.2.0(December 17,2025)
   - Added `SnowflakeCursor.stats` property to expose granular DML statistics (rows inserted, deleted, updated, and duplicates) for operations like CTAS where `rowcount` is insufficient.
   - Added support for injecting SPCS service identifier token (`SPCS_TOKEN`) into login requests when present in SPCS containers.
diff --git a/src/snowflake/connector/network.py b/src/snowflake/connector/network.py
index 4a8faa3ab..03760e8d8 100644
--- a/src/snowflake/connector/network.py
+++ b/src/snowflake/connector/network.py
@@ -308,6 +308,33 @@ def default(self, o):
         return super().default(o)
 
 
+class RetryCtx(TimeoutBackoffCtx):
+    def __init__(
+        self,
+        _include_retry_params: bool = False,
+        _include_retry_reason: bool = False,
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self.retry_reason = 0
+        self._include_retry_params = _include_retry_params
+        self._include_retry_reason = _include_retry_reason
+
+    def add_retry_params(self, full_url: str) -> str:
+        if self._include_retry_params and self.current_retry_count > 0:
+            retry_params = {
+                "clientStartTime": self._start_time_millis,
+                "retryCount": self.current_retry_count,
+            }
+            if self._include_retry_reason:
+                retry_params.update({"retryReason": self.retry_reason})
+            suffix = urlencode(retry_params)
+            sep = "&" if urlparse(full_url).query else "?"
+            return full_url + sep + suffix
+        else:
+            return full_url
+
+
 class SnowflakeRestful:
     """Snowflake Restful class."""
 
@@ -828,33 +855,6 @@ def fetch(
         **kwargs,
     ) -> dict[Any, Any]:
         """Carry out API request with session management."""
-
-        class RetryCtx(TimeoutBackoffCtx):
-            def __init__(
-                self,
-                _include_retry_params: bool = False,
-                _include_retry_reason: bool = False,
-                **kwargs,
-            ) -> None:
-                super().__init__(**kwargs)
-                self.retry_reason = 0
-                self._include_retry_params = _include_retry_params
-                self._include_retry_reason = _include_retry_reason
-
-            def add_retry_params(self, full_url: str) -> str:
-                if self._include_retry_params and self.current_retry_count > 0:
-                    retry_params = {
-                        "clientStartTime": self._start_time_millis,
-                        "retryCount": self.current_retry_count,
-                    }
-                    if self._include_retry_reason:
-                        retry_params.update({"retryReason": self.retry_reason})
-                    suffix = urlencode(retry_params)
-                    sep = "&" if urlparse(full_url).query else "?"
-                    return full_url + sep + suffix
-                else:
-                    return full_url
-
         include_retry_reason = self._connection._enable_retry_reason_in_query_response
         include_retry_params = kwargs.pop("_include_retry_params", False)