How to Use Socks5 Proxies in aiohttp

SOCKS5 proxies allow you to route network traffic through intermediary servers to improve privacy and access to geo-restricted content.

In this tutorial, you’ll learn how to integrate SOCKS5 proxies with aiohttp.

 

 

Configure aiohttp ClientSession with SOCKS5

To use SOCKS5 proxies with aiohttp, you’ll need to install the aiohttp-socks library.

This extension adds SOCKS proxy support to aiohttp.

Install the required libraries using pip:

!pip install aiohttp aiohttp-socks

Now, set up a ClientSession with a SOCKS5 proxy:

import aiohttp
from aiohttp_socks import ProxyConnector
import asyncio
async def main():
    connector = ProxyConnector.from_url('socks5://user:pass@127.0.0.1:9150')
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get('http://httpbin.org/ip') as response:
            print(await response.text())
asyncio.run(main())

Output:

{
  "origin": "185.220.101.34"
}

The output shows an IP address different from your actual IP to confirm that the request was routed through the SOCKS5 proxy.

 

Make Requests Through SOCKS5 Proxy

GET requests using SOCKS5

Perform a GET request through the configured SOCKS5 proxy:

async def get_request():
    connector = ProxyConnector.from_url('socks5://user:pass@127.0.0.1:9150')
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get('http://httpbin.org/get') as response:
            print(await response.json())
asyncio.run(get_request())

Output:

{'args': {}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Host': 'httpbin.org', 'User-Agent': 'Python/3.11 aiohttp/3.10.5', 'X-Amzn-Trace-Id': 'Root=1-66d538ad-3d0c126153ae0f261b81d67f'}, 'origin': '98.175.31.195', 'url': 'http://httpbin.org/get'}

The response shows that the request was successfully routed through the SOCKS5 proxy.

POST requests using SOCKS5

You can send a POST request through the SOCKS5 proxy like this:

async def post_request():
    connector = ProxyConnector.from_url('socks5://user:pass@127.0.0.1:9150')
    async with aiohttp.ClientSession(connector=connector) as session:
        data = {'key': 'value'}
        async with session.post('http://httpbin.org/post', data=data) as response:
            print(await response.json())
asyncio.run(post_request())

Output:

{'args': {}, 'data': '', 'files': {}, 'form': {'key': 'value'}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Content-Length': '9', 'Content-Type': 'application/x-www-form-urlencoded', 'Host': 'httpbin.org', 'User-Agent': 'Python/3.11 aiohttp/3.10.5', 'X-Amzn-Trace-Id': 'Root=1-66d53916-60cb52d5685f6c2f5202d04e'}, 'json': None, 'origin': '98.175.31.195', 'url': 'http://httpbin.org/post'}

The response confirms that the POST request was sent through the SOCKS5 proxy, with the form data correctly transmitted.

Handling other HTTP methods

Use other HTTP methods like PUT or DELETE through the SOCKS5 proxy:

async def put_request():
    connector = ProxyConnector.from_url('socks5://user:pass@127.0.0.1:9150')
    async with aiohttp.ClientSession(connector=connector) as session:
        data = {'updated': 'information'}
        async with session.put('http://httpbin.org/put', json=data) as response:
            print(await response.json())
asyncio.run(put_request())

Output:

{'args': {}, 'data': '{"updated": "information"}', 'files': {}, 'form': {}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Content-Length': '26', 'Content-Type': 'application/json', 'Host': 'httpbin.org', 'User-Agent': 'Python/3.11 aiohttp/3.10.5', 'X-Amzn-Trace-Id': 'Root=1-66d5395e-610b735b48274fb06da41acb'}, 'json': {'updated': 'information'}, 'origin': '98.175.31.195', 'url': 'http://httpbin.org/put'}

The response shows that the PUT request was successfully sent through the SOCKS5 proxy, with the JSON data correctly transmitted.

 

Proxy Chaining with aiohttp

You can chain multiple SOCKS5 proxies for improved anonymity:

async def chain_proxies():
    proxies = [
        'socks5://user1:pass1@98.175.31.195:4145',
        'socks5://user2:pass2@72.195.101.99:4145'
    ]
    for proxy in proxies:
        connector = ProxyConnector.from_url(proxy)
        async with aiohttp.ClientSession(connector=connector) as session:
            async with session.get('http://httpbin.org/ip') as response:
                print(f"Using proxy: {proxy}")
                print(await response.text())
asyncio.run(chain_proxies())

Output:

Using proxy: socks5://user1:pass1@98.175.31.195:4145
{
  "origin": "98.175.31.195"
}

Using proxy: socks5://user2:pass2@72.195.101.99:4145
{
  "origin": "72.195.101.99"
}

The output shows that requests were routed through different proxies sequentially.

Manage connection pools for multiple proxies

You can create separate connection pools for each proxy to manage resources efficiently:

async def manage_connection_pools():
    proxies = [
        'socks5://user1:pass1@98.175.31.195:4145',
        'socks5://user2:pass2@72.195.101.99:4145'
    ]
    connectors = [ProxyConnector.from_url(proxy) for proxy in proxies]
    sessions = [aiohttp.ClientSession(connector=connector) for connector in connectors]

    for session in sessions:
        async with session.get('http://httpbin.org/ip') as response:
            print(await response.text())

    for session in sessions:
        await session.close()
asyncio.run(manage_connection_pools())

Output:

{
  "origin": "98.175.31.195"
}

{
  "origin": "72.195.101.99"
}

This method creates separate connection pools for each proxy so you can send parallel requests through different proxies.

 

Streaming Responses Using SOCKS5 Proxies

You can handle streaming responses through a SOCKS5 proxy:

async def stream_response():
    connector = ProxyConnector.from_url('socks5://user:pass@127.0.0.1:9150')
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get('http://httpbin.org/stream/3') as response:
            async for line in response.content:
                print(line.decode().strip())
asyncio.run(stream_response())

Output:

{"url": "http://httpbin.org/stream/3", "args": {}, "headers": {"Host": "httpbin.org", "X-Amzn-Trace-Id": "Root=1-66d53b3b-1db9c8d47c4baff47a41cfe6", "Accept": "*/*", "Accept-Encoding": "gzip, deflate", "User-Agent": "Python/3.11 aiohttp/3.10.5"}, "origin": "98.175.31.195", "id": 0}
{"url": "http://httpbin.org/stream/3", "args": {}, "headers": {"Host": "httpbin.org", "X-Amzn-Trace-Id": "Root=1-66d53b3b-1db9c8d47c4baff47a41cfe6", "Accept": "*/*", "Accept-Encoding": "gzip, deflate", "User-Agent": "Python/3.11 aiohttp/3.10.5"}, "origin": "98.175.31.195", "id": 1}
{"url": "http://httpbin.org/stream/3", "args": {}, "headers": {"Host": "httpbin.org", "X-Amzn-Trace-Id": "Root=1-66d53b3b-1db9c8d47c4baff47a41cfe6", "Accept": "*/*", "Accept-Encoding": "gzip, deflate", "User-Agent": "Python/3.11 aiohttp/3.10.5"}, "origin": "98.175.31.195", "id": 2}

This example shows how to handle streaming responses through a SOCKS5 proxy.

 

Rate limiting Proxy Requests

You can implement rate limiting for requests through SOCKS5 proxies to prevent overuse of the proxies:

import time
from collections import defaultdict
class RateLimiter:
    def __init__(self, rate_limit):
        self.rate_limit = rate_limit
        self.tokens = defaultdict(lambda: self.rate_limit)
        self.updated_at = defaultdict(float)

    async def wait(self, proxy):
        while self.tokens[proxy] < 1:
            self.add_new_tokens(proxy)
            await asyncio.sleep(0.1)
        self.tokens[proxy] -= 1

    def add_new_tokens(self, proxy):
        now = time.time()
        time_since_update = now - self.updated_at[proxy]
        new_tokens = time_since_update * self.rate_limit
        if new_tokens > 1:
            self.tokens[proxy] = min(self.tokens[proxy] + new_tokens, self.rate_limit)
            self.updated_at[proxy] = now

async def rate_limited_requests():
    proxies = [
        'socks5://user1:pass1@98.175.31.195:4145',
        'socks5://user2:pass2@72.195.101.99:4145'
    ]
    rate_limiter = RateLimiter(rate_limit=2)  # 2 requests per second
    for proxy in proxies:
        connector = ProxyConnector.from_url(proxy)
        async with aiohttp.ClientSession(connector=connector) as session:
            for _ in range(5):
                await rate_limiter.wait(proxy)
                async with session.get('http://httpbin.org/ip') as response:
                    print(f"Proxy: {proxy}, Response: {await response.text()}")
asyncio.run(rate_limited_requests())

Output:

Proxy: socks5://user1:pass1@98.175.31.195:4145, Response: {
  "origin": "98.175.31.195"
}

Proxy: socks5://user1:pass1@98.175.31.195:4145, Response: {
  "origin": "98.175.31.195"
}

Proxy: socks5://user1:pass1@98.175.31.195:4145, Response: {
  "origin": "98.175.31.195"
}

Proxy: socks5://user1:pass1@98.175.31.195:4145, Response: {
  "origin": "98.175.31.195"
}

Proxy: socks5://user1:pass1@98.175.31.195:4145, Response: {
  "origin": "98.175.31.195"
}

Proxy: socks5://user2:pass2@72.195.101.99:4145, Response: {
  "origin": "72.195.101.99"
}

Proxy: socks5://user2:pass2@72.195.101.99:4145, Response: {
  "origin": "72.195.101.99"
}

Proxy: socks5://user2:pass2@72.195.101.99:4145, Response: {
  "origin": "72.195.101.99"
}

Proxy: socks5://user2:pass2@72.195.101.99:4145, Response: {
  "origin": "72.195.101.99"
}

Proxy: socks5://user2:pass2@72.195.101.99:4145, Response: {
  "origin": "72.195.101.99"
}

The output shows that requests through each proxy are rate-limited to 2 per second.

 

Geolocation-based Proxy Selection

Select SOCKS5 proxies based on desired geolocation:

import random
PROXY_LIST = {
    'US': ['socks5://user1:pass1@98.175.31.195:4145', 'socks5://user2:pass2@72.195.101.99:4145'],
    'CA': ['socks5://user3:pass3@51.161.119.166:63445', 'socks5://user4:pass4@192.111.129.145:16894'],
    'RU': ['socks5://user5:pass5@89.104.108.244:1080', 'socks5://user6:pass6@178.207.10.126:1080']
}
async def geo_based_request(region):
    if region not in PROXY_LIST:
        raise ValueError(f"No proxies available for region: {region}")

    proxy = random.choice(PROXY_LIST[region])
    connector = ProxyConnector.from_url(proxy)

    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get('http://httpbin.org/ip') as response:
            print(f"Region: {region}, Proxy: {proxy}")
            print(await response.text())
async def main():
    regions = ['US', 'CA', 'RU']
    tasks = [geo_based_request(region) for region in regions]
    await asyncio.gather(*tasks)
asyncio.run(main())

Output:

Region: CA, Proxy: socks5://user4:pass4@192.111.129.145:16894
{
  "origin": "192.111.129.145"
}

Region: US, Proxy: socks5://user1:pass1@98.175.31.195:4145
{
  "origin": "98.175.31.195"
}
Region: RU, Proxy: socks5://user5:pass5@89.104.108.244:1080
{
  "origin": "89.104.108.244"
}

This code shows you how to select proxies based on desired geographic locations.

It’s useful for accessing region-specific content or testing from different global locations.