Python
使用 requests、httpx 或 aiohttp 从 Python 连接到 RaxyProxy。
requests
在 Python 中使用代理最简单的方式。除 pip install requests.
basic.py
import requests
PROXY = "http://USERNAME:PASSWORD@proxy.raxyproxy.com:823"
proxies = {
"http": PROXY,
"https": PROXY,
}
r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=15)
print(r.json()) # {"origin": "203.0.113.42"}地理定向
geo.py
import requests
# Route through US IPs only (free, no 2× surcharge)
proxies_us = {
"http": "http://USERNAME__cr.us:PASSWORD@proxy.raxyproxy.com:823",
"https": "http://USERNAME__cr.us:PASSWORD@proxy.raxyproxy.com:823",
}
# City-level (Los Angeles) — consumes 2× traffic for residential/mobile/datacenter
proxies_la = {
"http": "http://USERNAME__cr.us;city.losangeles:PASSWORD@proxy.raxyproxy.com:823",
"https": "http://USERNAME__cr.us;city.losangeles:PASSWORD@proxy.raxyproxy.com:823",
}
r = requests.get("https://httpbin.org/ip", proxies=proxies_us)
print(r.json())粘性会话
sticky.py
import requests
# sessid sticky — same IP for ~30 min on standard port
SESSION_ID = "my-workflow-abc123"
proxy_url = f"http://USERNAME__sessid.{SESSION_ID}:PASSWORD@proxy.raxyproxy.com:823"
proxies = {"http": proxy_url, "https": proxy_url}
r1 = requests.get("https://httpbin.org/ip", proxies=proxies)
r2 = requests.get("https://httpbin.org/ip", proxies=proxies)
print(r1.json(), r2.json()) # same IP both times
# Port-based sticky — port 10001 always returns the same IP
proxies_sticky_port = {
"http": "http://USERNAME:PASSWORD@proxy.raxyproxy.com:10001",
"https": "http://USERNAME:PASSWORD@proxy.raxyproxy.com:10001",
}
r3 = requests.get("https://httpbin.org/ip", proxies=proxies_sticky_port)
print(r3.json())httpx (async)
pip install httpx — 支持异步并内置代理支持。
httpx_example.py
import asyncio
import httpx
PROXY = "http://USERNAME:PASSWORD@proxy.raxyproxy.com:823"
async def fetch(url: str) -> dict:
async with httpx.AsyncClient(proxy=PROXY, timeout=15) as client:
r = await client.get(url)
return r.json()
async def main():
# Run 10 concurrent requests — each gets a different IP
tasks = [fetch("https://httpbin.org/ip") for _ in range(10)]
results = await asyncio.gather(*tasks)
for r in results:
print(r["origin"])
asyncio.run(main())aiohttp
pip install aiohttp
aiohttp_example.py
import asyncio
import aiohttp
PROXY = "http://proxy.raxyproxy.com:823"
AUTH = aiohttp.BasicAuth("USERNAME", "PASSWORD")
async def fetch(session: aiohttp.ClientSession, url: str) -> str:
async with session.get(url, proxy=PROXY, proxy_auth=AUTH) as r:
data = await r.json()
return data["origin"]
async def main():
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, "https://httpbin.org/ip") for _ in range(5)]
results = await asyncio.gather(*tasks)
print(results)
asyncio.run(main())会话池模式
管理粘性代理槽池,用于并行独立会话。
pool.py
import requests
from concurrent.futures import ThreadPoolExecutor
HOST = "proxy.raxyproxy.com"
USER = "USERNAME"
PASS = "PASSWORD"
def make_proxy(port: int) -> dict:
"""Each port 10001–10010 is a dedicated sticky slot."""
url = f"http://{USER}:{PASS}@{HOST}:{port}"
return {"http": url, "https": url}
def scrape(task: tuple[int, str]) -> dict:
slot, url = task
proxies = make_proxy(10000 + slot)
r = requests.get(url, proxies=proxies, timeout=15)
return {"slot": slot, "ip": r.json()["origin"], "url": url}
urls = [f"https://example.com/page/{i}" for i in range(10)]
tasks = list(enumerate(urls))
with ThreadPoolExecutor(max_workers=10) as pool:
results = list(pool.map(scrape, tasks))
for r in results:
print(f"Slot {r['slot']} (IP {r['ip']}): {r['url']}")对于大规模抓取,请将轮换代理(端口 823)与重试机制结合使用。请参阅 Troubleshooting for common error codes and how to handle them.