tessl/pypi-python-socks

Proxy (SOCKS4, SOCKS5, HTTP CONNECT) client for Python

—

Pending

Overview

Eval results

Files

V2 Enhanced API

Name: tessl/pypi-python-socks
Author: tessl

Enhanced proxy implementations with improved error handling, SSL support, and additional configuration options. The V2 API provides the latest features and is recommended for new applications. Available for both synchronous and asynchronous usage patterns across multiple frameworks.

V2 API Imports

# Synchronous V2
from python_socks.sync.v2 import Proxy as SyncProxyV2, ProxyChain as SyncProxyChainV2

# Asyncio V2  
from python_socks.async_.asyncio.v2 import Proxy as AsyncioProxyV2, ProxyChain as AsyncioProxyChainV2

# Trio V2
from python_socks.async_.trio.v2 import Proxy as TrioProxyV2, ProxyChain as TrioProxyChainV2

# AnyIO V2
from python_socks.async_.anyio.v2 import Proxy as AnyioProxyV2, ProxyChain as AnyioProxyChainV2

Capabilities

Synchronous V2 Proxy

Enhanced synchronous proxy with improved features and error handling.

import ssl
from typing import Optional

class SyncProxy:
    def __init__(
        self,
        proxy_type: ProxyType,
        host: str,
        port: int,
        username: Optional[str] = None,
        password: Optional[str] = None,
        rdns: Optional[bool] = None,
        proxy_ssl: Optional[ssl.SSLContext] = None,
        forward: Optional['SyncProxy'] = None
    ): ...
    
    def connect(
        self,
        dest_host: str,
        dest_port: int,
        dest_ssl: Optional[ssl.SSLContext] = None,
        timeout: Optional[float] = None,
        **kwargs
    ) -> 'SyncSocketStream': ...
    
    @property
    def proxy_host(self) -> str:
        """Get proxy host address."""
        ...
        
    @property  
    def proxy_port(self) -> int:
        """Get proxy port number."""
        ...
    
    @classmethod
    def create(cls, *args, **kwargs) -> 'SyncProxy':
        """Create proxy instance (deprecated, use __init__ directly)."""
        ...
    
    @classmethod
    def from_url(cls, url: str, **kwargs) -> 'SyncProxy':
        """Create proxy instance from URL."""
        ...

Enhanced Features:

Built-in SSL/TLS transport support via proxy_ssl and dest_ssl parameters
Proxy chaining support with forward parameter
Returns enhanced SyncSocketStream instead of raw socket
Better error reporting and diagnostics
Enhanced connection pooling compatibility
Optimized performance characteristics

V2-Specific Parameters:

proxy_ssl: Optional SSL context for connecting to the proxy server over TLS
forward: Optional proxy to forward connections through (proxy chaining)
dest_ssl: Optional SSL context for wrapping the destination connection

Usage:

import ssl
from python_socks.sync.v2 import Proxy

# Create V2 proxy
proxy = Proxy.from_url('socks5://user:pass@proxy.example.com:1080')

# Connect with enhanced features
sock = proxy.connect('secure-api.example.com', 443, timeout=30)

# Enhanced SSL integration
context = ssl.create_default_context()
secure_sock = context.wrap_socket(
    sock,
    server_hostname='secure-api.example.com'
)

# Make secure request
request = b'GET /api/v1/data HTTP/1.1\r\nHost: secure-api.example.com\r\n\r\n'
secure_sock.send(request)
response = secure_sock.recv(4096)
secure_sock.close()

SyncSocketStream Class

Enhanced socket wrapper returned by V2 proxy connections.

class SyncSocketStream:
    def __init__(self, sock: Union[socket.socket, ssl.SSLSocket]): ...
    
    def write_all(self, data: bytes) -> None:
        """Write all data to the socket."""
        ...
        
    def read(self, max_bytes: int = 65536) -> bytes:
        """Read up to max_bytes from the socket."""
        ...
        
    def read_exact(self, n: int) -> bytes:
        """Read exactly n bytes from the socket."""
        ...
        
    def close(self) -> None:
        """Close the underlying socket."""
        ...
        
    @property
    def socket(self) -> Union[socket.socket, ssl.SSLSocket]:
        """Get the underlying socket object."""
        ...

SyncSocketStream Features:

Consistent interface across different socket types (plain, SSL, custom transports)
Convenient methods for common I/O operations
Better error handling and resource management
Compatible with existing socket operations via .socket property

Usage:

from python_socks.sync.v2 import Proxy

proxy = Proxy.from_url('socks5://proxy:1080')
stream = proxy.connect('example.com', 80)

# Use enhanced stream methods
stream.write_all(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')
response = stream.read(4096)

# Or access underlying socket if needed
sock = stream.socket
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

stream.close()

Synchronous V2 Proxy Chain

Enhanced proxy chaining with better error handling and performance.

class ProxyChain:
    def __init__(self, proxies: Iterable[SyncProxy]): ...
    
    def connect(
        self,
        dest_host: str,
        dest_port: int,
        timeout: Optional[float] = None
    ) -> socket.socket: ...

Enhanced Chain Features:

Better intermediate connection management
Improved error propagation
Enhanced timeout handling
Optimized chain traversal

Usage:

from python_socks.sync.v2 import Proxy, ProxyChain

# Create enhanced proxy chain
proxy1 = Proxy.from_url('socks5://first-proxy:1080')
proxy2 = Proxy.from_url('http://second-proxy:8080')
proxy3 = Proxy.from_url('socks4://third-proxy:1080')

chain = ProxyChain([proxy1, proxy2, proxy3])

# Connect through enhanced chain
sock = chain.connect('destination.example.com', 80, timeout=60)

# Use the connection
sock.send(b'GET / HTTP/1.1\r\nHost: destination.example.com\r\n\r\n')
response = sock.recv(4096)
print(response.decode())
sock.close()

Asyncio V2 Proxy

Enhanced asyncio proxy implementation with advanced async features.

class AsyncioProxy:
    def __init__(
        self,
        proxy_type: ProxyType,
        host: str,
        port: int,
        username: Optional[str] = None,
        password: Optional[str] = None,
        rdns: Optional[bool] = None,
        loop: asyncio.AbstractEventLoop | None = None
    ): ...
    
    async def connect(
        self,
        dest_host: str,
        dest_port: int,
        timeout: Optional[float] = None,
        **kwargs
    ) -> socket.socket: ...
    
    @classmethod
    def create(cls, *args, **kwargs) -> 'AsyncioProxy': ...
    
    @classmethod
    def from_url(cls, url: str, **kwargs) -> 'AsyncioProxy': ...

Enhanced Async Features:

Better cancellation support
Improved timeout mechanisms
Enhanced connection pooling
Optimized for asyncio 3.11+ features

Usage:

import asyncio
import ssl
from python_socks.async_.asyncio.v2 import Proxy

async def enhanced_async_example():
    # Create V2 asyncio proxy
    proxy = Proxy.from_url('socks5://proxy.example.com:1080')
    
    # Enhanced async connect
    sock = await proxy.connect('api.example.com', 443, timeout=30)
    
    # Use with enhanced asyncio features
    reader, writer = await asyncio.open_connection(
        host=None,
        port=None,
        sock=sock,
        ssl=ssl.create_default_context(),
        server_hostname='api.example.com'
    )
    
    # Enhanced request handling
    request = b'GET /v2/endpoint HTTP/1.1\r\nHost: api.example.com\r\n\r\n'
    writer.write(request)
    await writer.drain()
    
    response = await reader.read(4096)
    print(response.decode())
    
    writer.close()
    await writer.wait_closed()

asyncio.run(enhanced_async_example())

Asyncio V2 Proxy Chain

Enhanced asynchronous proxy chaining for asyncio.

class ProxyChain:
    def __init__(self, proxies: Iterable[AsyncioProxy]): ...
    
    async def connect(
        self,
        dest_host: str,
        dest_port: int,
        timeout: Optional[float] = None
    ) -> socket.socket: ...

Usage:

import asyncio
from python_socks.async_.asyncio.v2 import Proxy, ProxyChain

async def enhanced_chain_example():
    # Create enhanced async proxy chain
    proxy1 = Proxy.from_url('socks5://proxy1.example.com:1080')
    proxy2 = Proxy.from_url('http://proxy2.example.com:8080')
    
    chain = ProxyChain([proxy1, proxy2])
    
    # Enhanced async chain connection
    sock = await chain.connect('destination.example.com', 80, timeout=45)
    
    # Use the connection
    reader, writer = await asyncio.open_connection(
        host=None, port=None, sock=sock
    )
    
    writer.write(b'GET / HTTP/1.1\r\nHost: destination.example.com\r\n\r\n')
    await writer.drain()
    
    response = await reader.read(4096)
    print(response.decode())
    
    writer.close()
    await writer.wait_closed()

asyncio.run(enhanced_chain_example())

Trio V2 Proxy

Enhanced trio proxy implementation with structured concurrency improvements.

class TrioProxy:
    def __init__(
        self,
        proxy_type: ProxyType,
        host: str,
        port: int,
        username: Optional[str] = None,
        password: Optional[str] = None,
        rdns: Optional[bool] = None
    ): ...
    
    async def connect(
        self,
        dest_host: str,
        dest_port: int,
        timeout: Optional[float] = None,
        **kwargs
    ) -> socket.socket: ...
    
    @classmethod
    def create(cls, *args, **kwargs) -> 'TrioProxy': ...
    
    @classmethod
    def from_url(cls, url: str, **kwargs) -> 'TrioProxy': ...

Enhanced Trio Features:

Better structured concurrency integration
Improved cancellation handling
Enhanced memory management
Optimized for trio's nursery patterns

Usage:

import trio
import ssl
from python_socks.async_.trio.v2 import Proxy

async def enhanced_trio_example():
    # Create V2 trio proxy
    proxy = Proxy.from_url('socks5://proxy.example.com:1080')
    
    # Enhanced trio connect
    sock = await proxy.connect('secure-api.example.com', 443)
    
    # Enhanced SSL integration with trio
    ssl_context = ssl.create_default_context()
    stream = trio.SSLStream(
        trio.SocketStream(sock),
        ssl_context,
        server_hostname='secure-api.example.com'
    )
    
    # Make secure request
    await stream.send_all(b'GET /api/data HTTP/1.1\r\nHost: secure-api.example.com\r\n\r\n')
    response = await stream.receive_some(4096)
    print(response.decode())
    
    await stream.aclose()

trio.run(enhanced_trio_example)

Trio V2 Proxy Chain

Enhanced proxy chaining for trio with better structured concurrency support.

class ProxyChain:
    def __init__(self, proxies: Iterable[TrioProxy]): ...
    
    async def connect(
        self,
        dest_host: str,
        dest_port: int,
        timeout: Optional[float] = None
    ) -> socket.socket: ...

Usage:

import trio
from python_socks.async_.trio.v2 import Proxy, ProxyChain

async def enhanced_trio_chain_example():
    # Create enhanced trio proxy chain
    proxy1 = Proxy.from_url('socks5://proxy1.example.com:1080')
    proxy2 = Proxy.from_url('http://proxy2.example.com:8080')
    
    chain = ProxyChain([proxy1, proxy2])
    
    # Enhanced chain connection
    sock = await chain.connect('example.com', 80)
    
    # Use with trio streams
    stream = trio.SocketStream(sock)
    
    await stream.send_all(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')
    response = await stream.receive_some(4096)
    print(response.decode())
    
    await stream.aclose()

trio.run(enhanced_trio_chain_example)

AnyIO V2 Proxy

Enhanced AnyIO proxy with cross-framework improvements.

class AnyioProxy:
    def __init__(
        self,
        proxy_type: ProxyType,
        host: str,
        port: int,
        username: Optional[str] = None,
        password: Optional[str] = None,
        rdns: Optional[bool] = None
    ): ...
    
    async def connect(
        self,
        dest_host: str,
        dest_port: int,
        timeout: Optional[float] = None,
        **kwargs
    ) -> socket.socket: ...
    
    @classmethod
    def create(cls, *args, **kwargs) -> 'AnyioProxy': ...
    
    @classmethod
    def from_url(cls, url: str, **kwargs) -> 'AnyioProxy': ...

Enhanced AnyIO Features:

Better cross-framework compatibility
Improved backend detection
Enhanced error handling across frameworks
Optimized performance for all supported backends

Usage:

import anyio
import ssl
from python_socks.async_.anyio.v2 import Proxy

async def enhanced_anyio_example():
    # Create V2 AnyIO proxy
    proxy = Proxy.from_url('socks5://proxy.example.com:1080')
    
    # Enhanced cross-framework connect
    sock = await proxy.connect('secure-api.example.com', 443)
    
    # Enhanced SSL support across frameworks
    ssl_context = ssl.create_default_context()
    stream = await anyio.connect_tcp(
        remote_host='secure-api.example.com',
        remote_port=443,
        sock=sock,
        tls=True,
        tls_hostname='secure-api.example.com'
    )
    
    # Make secure request
    await stream.send(b'GET /v2/data HTTP/1.1\r\nHost: secure-api.example.com\r\n\r\n')
    response = await stream.receive(4096)
    print(response.decode())
    
    await stream.aclose()

# Works with any supported backend
anyio.run(enhanced_anyio_example, backend='asyncio')

AnyIO V2 Proxy Chain

Enhanced proxy chaining with cross-framework compatibility.

class ProxyChain:
    def __init__(self, proxies: Iterable[AnyioProxy]): ...
    
    async def connect(
        self,
        dest_host: str,
        dest_port: int,
        timeout: Optional[float] = None
    ) -> socket.socket: ...

Usage:

import anyio
from python_socks.async_.anyio.v2 import Proxy, ProxyChain

async def enhanced_anyio_chain_example():
    # Create enhanced AnyIO proxy chain
    proxy1 = Proxy.from_url('socks5://proxy1.example.com:1080')
    proxy2 = Proxy.from_url('http://proxy2.example.com:8080')
    
    chain = ProxyChain([proxy1, proxy2])
    
    # Enhanced chain connection
    sock = await chain.connect('api.example.com', 443)
    
    # Cross-framework usage
    stream = anyio.SocketStream(sock)
    
    await stream.send(b'GET /api/v2/endpoint HTTP/1.1\r\nHost: api.example.com\r\n\r\n')
    response = await stream.receive(4096)
    print(response.decode())
    
    await stream.aclose()

# Test with different backends
async def test_backends():
    for backend in ['asyncio', 'trio']:
        print(f"Testing with {backend}")
        await anyio.run(enhanced_anyio_chain_example, backend=backend)

anyio.run(test_backends, backend='asyncio')

V2 API Improvements

Enhanced Error Handling

The V2 API provides more detailed error information and better error categorization:

from python_socks import ProxyError
from python_socks.sync.v2 import Proxy

try:
    proxy = Proxy.from_url('socks5://invalid-proxy:1080')
    sock = proxy.connect('example.com', 80)
except ProxyError as e:
    # V2 provides enhanced error details
    print(f"Error: {e}")
    print(f"Error code: {e.error_code}")
    print(f"Protocol-specific info: {getattr(e, 'protocol_info', 'N/A')}")

Performance Optimizations

V2 implementations include various performance improvements:

Optimized connection establishment
Better memory usage patterns
Reduced syscall overhead
Enhanced connection pooling support

SSL/TLS Enhancements

Improved SSL/TLS support across all V2 implementations:

import ssl
from python_socks.sync.v2 import Proxy

# Enhanced SSL configuration support
proxy = Proxy.from_url('socks5://proxy.example.com:1080')
sock = proxy.connect('secure-api.example.com', 443)

# V2 provides better SSL context integration
context = ssl.create_default_context()
context.check_hostname = True
context.verify_mode = ssl.CERT_REQUIRED

secure_sock = context.wrap_socket(
    sock,
    server_hostname='secure-api.example.com'
)

Migration from V1 to V2

The V2 API maintains backward compatibility while adding enhancements:

# V1 usage
from python_socks.sync import Proxy as ProxyV1

# V2 usage (recommended)
from python_socks.sync.v2 import Proxy as ProxyV2

# Same interface, enhanced implementation
proxy_v1 = ProxyV1.from_url('socks5://proxy.example.com:1080')
proxy_v2 = ProxyV2.from_url('socks5://proxy.example.com:1080')

# Both work the same way, V2 has better performance and features
sock_v1 = proxy_v1.connect('example.com', 80)
sock_v2 = proxy_v2.connect('example.com', 80)  # Recommended

V2 Best Practices

Use V2 for New Applications

Always use V2 implementations for new projects:

# Recommended imports for new applications
from python_socks.sync.v2 import Proxy, ProxyChain
from python_socks.async_.asyncio.v2 import Proxy as AsyncProxy

Enhanced Connection Management

V2 provides better connection lifecycle management:

from contextlib import asynccontextmanager
from python_socks.async_.asyncio.v2 import Proxy

@asynccontextmanager
async def proxy_connection_v2(proxy_url, dest_host, dest_port):
    """Enhanced V2 connection context manager."""
    proxy = Proxy.from_url(proxy_url)
    sock = None
    try:
        sock = await proxy.connect(dest_host, dest_port, timeout=30)
        yield sock
    finally:
        if sock:
            sock.close()

# Usage with enhanced features
async with proxy_connection_v2('socks5://proxy:1080', 'example.com', 80) as sock:
    # V2 provides enhanced socket configuration
    reader, writer = await asyncio.open_connection(
        host=None, port=None, sock=sock
    )
    # ... use connection

Install with Tessl CLI