Add type hints to urllib3.contrib.socks
MODIFY
noxfile.py
noxfile.py
@@ -9,6 +9,7 @@ import nox
# ignored.
TYPED_FILES = {
"src/urllib3/contrib/__init__.py",
+ "src/urllib3/contrib/socks.py",
"src/urllib3/__init__.py",
"src/urllib3/connection.py",
"src/urllib3/connectionpool.py",

MODIFY
src/urllib3/connection.py
src/urllib3/connection.py
@@ -211,7 +211,7 @@ class HTTPConnection(_HTTPConnection):
)
except OSError as e:
- raise NewConnectionError(self, f"Failed to establish a new connection: {e}") # type: ignore
+ raise NewConnectionError(self, f"Failed to establish a new connection: {e}")
return conn

MODIFY
src/urllib3/contrib/socks.py
src/urllib3/contrib/socks.py
@@ -39,7 +39,7 @@ with the proxy:
"""
try:
- import socks
+ import socks # type: ignore
except ImportError:
import warnings
@@ -56,6 +56,7 @@ except ImportError:
raise
from socket import timeout as SocketTimeout
+from typing import Any, Dict, Mapping, Optional
from ..connection import HTTPConnection, HTTPSConnection
from ..connectionpool import HTTPConnectionPool, HTTPSConnectionPool
@@ -66,7 +67,22 @@ from ..util.url import parse_url
try:
import ssl
except ImportError:
- ssl = None
+ ssl = None # type: ignore
+
+try:
+ from typing import TypedDict
+
+ class _TYPE_SOCKS_OPTIONS(TypedDict):
+ socks_version: int
+ proxy_host: Optional[str]
+ proxy_port: Optional[str]
+ username: Optional[str]
+ password: Optional[str]
+ rdns: bool
+
+
+except ImportError: # Python 3.7
+ _TYPE_SOCKS_OPTIONS = Dict[str, Any] # type: ignore
class SOCKSConnection(HTTPConnection):
@@ -74,15 +90,17 @@ class SOCKSConnection(HTTPConnection):
A plain-text HTTP connection that connects via a SOCKS proxy.
"""
- def __init__(self, *args, **kwargs):
- self._socks_options = kwargs.pop("_socks_options")
+ def __init__(
+ self, _socks_options: _TYPE_SOCKS_OPTIONS, *args: Any, **kwargs: Any
+ ) -> None:
+ self._socks_options = _socks_options
super().__init__(*args, **kwargs)
- def _new_conn(self):
+ def _new_conn(self) -> "socks.socksocket":
"""
Establish a new connection via the SOCKS proxy.
"""
- extra_kw = {}
+ extra_kw: Dict[str, Any] = {}
if self.source_address:
extra_kw["source_address"] = self.source_address
@@ -162,12 +180,12 @@ class SOCKSProxyManager(PoolManager):
def __init__(
self,
- proxy_url,
- username=None,
- password=None,
- num_pools=10,
- headers=None,
- **connection_pool_kw,
+ proxy_url: str,
+ username: Optional[str] = None,
+ password: Optional[str] = None,
+ num_pools: int = 10,
+ headers: Optional[Mapping[str, str]] = None,
+ **connection_pool_kw: Any,
):
parsed = parse_url(proxy_url)

MODIFY
src/urllib3/exceptions.py
src/urllib3/exceptions.py
+import warnings
from http.client import IncompleteRead as httplib_IncompleteRead
from typing import TYPE_CHECKING, Callable, List, Optional, Tuple, Union
if TYPE_CHECKING:
from email.errors import MessageDefect
+ from urllib3.connection import HTTPConnection
from urllib3.connectionpool import ConnectionPool
from urllib3.response import HTTPResponse
from urllib3.util.retry import Retry
@@ -160,10 +162,25 @@ class ConnectTimeoutError(TimeoutError):
pass
-class NewConnectionError(ConnectTimeoutError, PoolError):
+class NewConnectionError(ConnectTimeoutError, HTTPError):
"""Raised when we fail to establish a new connection. Usually ECONNREFUSED."""
- pass
+ conn: "HTTPConnection"
+
+ def __init__(self, conn: "HTTPConnection", message: str) -> None:
+ self.conn = conn
+ super().__init__(f"{conn}: {message}")
+
+ @property
+ def pool(self) -> "HTTPConnection":
+ warnings.warn(
+ "The 'pool' property is deprecated and will be removed "
+ "in a later urllib3 v2.x release. use 'conn' instead.",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+
+ return self.conn
class EmptyPoolError(PoolError):

MODIFY
test/test_exceptions.py
test/test_exceptions.py
@@ -2,6 +2,7 @@ import pickle
import pytest
+from urllib3.connection import HTTPConnection
from urllib3.connectionpool import HTTPConnectionPool
from urllib3.exceptions import (
ClosedPoolError,
@@ -12,6 +13,7 @@ from urllib3.exceptions import (
HTTPError,
LocationParseError,
MaxRetryError,
+ NewConnectionError,
ReadTimeoutError,
)
@@ -45,3 +47,17 @@ class TestFormat:
assert "defects" in str(hpe)
assert "unparsed_data" in str(hpe)
+
+
+class TestNewConnectionError:
+ def test_pool_property_deprecation_warning(self):
+ err = NewConnectionError(HTTPConnection("localhost"), "test")
+ with pytest.warns(DeprecationWarning) as record:
+ err.pool
+
+ assert err.pool is err.conn
+ msg = (
+ "The 'pool' property is deprecated and will be removed "
+ "in a later urllib3 v2.x release. use 'conn' instead."
+ )
+ assert record[0].message.args[0] == msg

Add type hints to urllib3.contrib.socks