Proxy connector for aiohttp supporting SOCKS4/5 and HTTP proxies with proxy chaining capabilities
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Legacy functionality maintained for backward compatibility. These APIs are deprecated and users should migrate to newer alternatives or use python-socks directly.
Legacy connection functions that provide direct proxy socket connections. These are deprecated in favor of using python-socks directly.
async def open_connection(
proxy_url=None,
host=None,
port=None,
*,
proxy_type=ProxyType.SOCKS5,
proxy_host='127.0.0.1',
proxy_port=1080,
username=None,
password=None,
rdns=True,
loop=None,
**kwargs
):
"""
DEPRECATED: Open connection through proxy.
Use https://github.com/romis2012/python-socks directly instead.
Args:
proxy_url (str, optional): Proxy URL string
host (str): Target hostname (required)
port (int): Target port (required)
proxy_type (ProxyType): Type of proxy (default: SOCKS5)
proxy_host (str): Proxy hostname (default: '127.0.0.1')
proxy_port (int): Proxy port (default: 1080)
username (str, optional): Proxy username
password (str, optional): Proxy password
rdns (bool): Remote DNS resolution (default: True)
loop: Event loop (optional)
**kwargs: Additional arguments passed to asyncio.open_connection
Returns:
Tuple[StreamReader, StreamWriter]: Connected stream objects
Raises:
ValueError: If host or port not specified
DeprecationWarning: Always emitted when called
"""
async def create_connection(
proxy_url=None,
protocol_factory=None,
host=None,
port=None,
*,
proxy_type=ProxyType.SOCKS5,
proxy_host='127.0.0.1',
proxy_port=1080,
username=None,
password=None,
rdns=True,
loop=None,
**kwargs
):
"""
DEPRECATED: Create connection through proxy.
Use https://github.com/romis2012/python-socks directly instead.
Args:
proxy_url (str, optional): Proxy URL string
protocol_factory: Protocol factory function (required)
host (str): Target hostname (required)
port (int): Target port (required)
proxy_type (ProxyType): Type of proxy (default: SOCKS5)
proxy_host (str): Proxy hostname (default: '127.0.0.1')
proxy_port (int): Proxy port (default: 1080)
username (str, optional): Proxy username
password (str, optional): Proxy password
rdns (bool): Remote DNS resolution (default: True)
loop: Event loop (optional)
**kwargs: Additional arguments passed to loop.create_connection
Returns:
Tuple[Transport, Protocol]: Connected transport and protocol
Raises:
ValueError: If protocol_factory, host, or port not specified
DeprecationWarning: Always emitted when called
"""Legacy connector classes and constants maintained for backward compatibility.
class SocksVer:
"""
DEPRECATED: Legacy SOCKS version constants.
Use ProxyType enum instead.
"""
SOCKS4 = 1
SOCKS5 = 2
class SocksConnector(ProxyConnector):
"""
DEPRECATED: Legacy SOCKS connector.
Use ProxyConnector instead.
Inherits all functionality from ProxyConnector but emits
deprecation warnings when used.
"""
def __init__(self, socks_ver=SocksVer.SOCKS5, **kwargs):
"""
Create legacy SOCKS connector.
Args:
socks_ver (int): SOCKS version using SocksVer constants
**kwargs: Arguments passed to ProxyConnector
Emits:
DeprecationWarning: Always emitted when instantiated
"""
@classmethod
def from_url(cls, url, **kwargs):
"""
Create legacy SOCKS connector from URL.
Args:
url (str): Proxy URL
**kwargs: Arguments passed to ProxyConnector
Returns:
SocksConnector: Legacy connector instance
Emits:
DeprecationWarning: Always emitted when called
"""Legacy exception names that alias to the current python-socks exceptions.
SocksError = ProxyError
"""DEPRECATED: Alias for ProxyError. Use ProxyError instead."""
SocksConnectionError = ProxyConnectionError
"""DEPRECATED: Alias for ProxyConnectionError. Use ProxyConnectionError instead."""# Old deprecated way
from aiohttp_socks import SocksConnector, SocksVer
connector = SocksConnector(
socks_ver=SocksVer.SOCKS5,
host='127.0.0.1',
port=1080
)
# New recommended way
from aiohttp_socks import ProxyConnector, ProxyType
connector = ProxyConnector(
proxy_type=ProxyType.SOCKS5,
host='127.0.0.1',
port=1080
)# Old deprecated way
from aiohttp_socks import open_connection
reader, writer = await open_connection(
host='example.com',
port=80,
proxy_host='127.0.0.1',
proxy_port=1080
)
# New recommended way
from python_socks.async_.asyncio import Proxy
proxy = Proxy.from_url('socks5://127.0.0.1:1080')
stream = await proxy.connect('example.com', 80)
reader, writer = stream.reader, stream.writer# Old deprecated way
from aiohttp_socks import SocksError, SocksConnectionError
try:
# proxy operations
pass
except SocksConnectionError as e:
print(f"Connection failed: {e}")
except SocksError as e:
print(f"Proxy error: {e}")
# New recommended way
from aiohttp_socks import ProxyError, ProxyConnectionError
try:
# proxy operations
pass
except ProxyConnectionError as e:
print(f"Connection failed: {e}")
except ProxyError as e:
print(f"Proxy error: {e}")Install with Tessl CLI
npx tessl i tessl/pypi-aiohttp-socks