Proxy connector for aiohttp supporting SOCKS4/5 and HTTP proxies with proxy chaining capabilities
npx @tessl/cli install tessl/pypi-aiohttp-socks@0.10.0A proxy connector for aiohttp supporting SOCKS4, SOCKS4a, SOCKS5, SOCKS5h, and HTTP tunneling proxies with proxy chaining capabilities. Built on python-socks for core proxy functionality with seamless aiohttp integration.
pip install aiohttp-socksfrom aiohttp_socks import ProxyConnector, ChainProxyConnector, ProxyType, ProxyInfoImport exceptions for error handling:
from aiohttp_socks import ProxyError, ProxyConnectionError, ProxyTimeoutErrorimport aiohttp
from aiohttp_socks import ProxyConnector, ProxyType
async def fetch_with_proxy(url):
# Create connector from URL
connector = ProxyConnector.from_url('socks5://user:password@127.0.0.1:1080')
# Or create connector with parameters
# connector = ProxyConnector(
# proxy_type=ProxyType.SOCKS5,
# host='127.0.0.1',
# port=1080,
# username='user',
# password='password',
# rdns=True
# )
async with aiohttp.ClientSession(connector=connector) as session:
async with session.get(url) as response:
return await response.text()
# Proxy chaining example
from aiohttp_socks import ChainProxyConnector
async def fetch_with_proxy_chain(url):
connector = ChainProxyConnector.from_urls([
'socks5://user:password@127.0.0.1:1080',
'socks4://127.0.0.1:1081',
'http://user:password@127.0.0.1:3128',
])
async with aiohttp.ClientSession(connector=connector) as session:
async with session.get(url) as response:
return await response.text()The library provides two main connector classes that extend aiohttp's TCPConnector:
Both connectors handle authentication, SSL contexts, and integrate seamlessly with aiohttp's connector architecture using a custom resolver to bypass DNS resolution.
Direct proxy connections supporting SOCKS4/4a/5/5h and HTTP tunneling with authentication and SSL support.
class ProxyConnector:
def __init__(
self,
host: str,
port: int,
proxy_type: ProxyType = ProxyType.SOCKS5,
username: Optional[str] = None,
password: Optional[str] = None,
rdns: Optional[bool] = None,
proxy_ssl: Optional[SSLContext] = None,
**kwargs: Any
) -> None: ...
@classmethod
def from_url(cls, url: str, **kwargs: Any) -> 'ProxyConnector': ...Sequential proxy routing through multiple proxy servers for enhanced anonymity and flexible network routing.
class ChainProxyConnector:
def __init__(self, proxy_infos: Iterable[ProxyInfo], **kwargs): ...
@classmethod
def from_urls(cls, urls: Iterable[str], **kwargs: Any) -> 'ChainProxyConnector': ...
class ProxyInfo(NamedTuple):
proxy_type: ProxyType
host: str
port: int
username: Optional[str] = None
password: Optional[str] = None
rdns: Optional[bool] = NoneComprehensive exception hierarchy for proxy connection failures and timeout handling.
class ProxyError(Exception): ...
class ProxyConnectionError(ProxyError): ...
class ProxyTimeoutError(ProxyError): ...Legacy functionality maintained for backward compatibility including deprecated utility functions and connector classes.
async def open_connection(...): ...
async def create_connection(...): ...
class SocksConnector(ProxyConnector): ...
class SocksVer: ...
SocksError = ProxyError
SocksConnectionError = ProxyConnectionError# Package constants
__title__: str = 'aiohttp-socks'
__version__: str = '0.10.1'
# Proxy types
from python_socks import ProxyType
# ProxyType.SOCKS4, ProxyType.SOCKS4a, ProxyType.SOCKS5, ProxyType.SOCKS5h, ProxyType.HTTP
# Type imports
from ssl import SSLContext
from typing import Optional, Any, Iterable, NamedTuple