A thread-safe proxy rotator with pluggable rotation strategies.
Supports:
- Cyclic rotation (default)
- Custom rotation strategies via callable
- Both string URLs and Playwright-style dict proxies
Initialize the proxy rotator.
| PARAMETER |
DESCRIPTION |
proxies
|
List of proxy URLs or Playwright-style proxy dicts.
- String format: "http://proxy1:8080" or "http://user:pass@proxy:8080"
- Dict format: {"server": "http://proxy:8080", "username": "user", "password": "pass"}
TYPE:
List[ProxyType]
|
strategy
|
Rotation strategy function. Takes (proxies, current_index) and returns (proxy, next_index). Defaults to cyclic_rotation.
TYPE:
RotationStrategy
DEFAULT:
cyclic_rotation
|
Source code in scrapling/engines/toolbelt/proxy_rotation.py
| def __init__(
self,
proxies: List[ProxyType],
strategy: RotationStrategy = cyclic_rotation,
):
"""
Initialize the proxy rotator.
:param proxies: List of proxy URLs or Playwright-style proxy dicts.
- String format: "http://proxy1:8080" or "http://user:pass@proxy:8080"
- Dict format: {"server": "http://proxy:8080", "username": "user", "password": "pass"}
:param strategy: Rotation strategy function. Takes (proxies, current_index) and returns (proxy, next_index). Defaults to cyclic_rotation.
"""
if not proxies:
raise ValueError("At least one proxy must be provided")
if not callable(strategy):
raise TypeError(f"strategy must be callable, got {type(strategy).__name__}")
self._strategy = strategy
self._lock = Lock()
# Validate and store proxies
self._proxies: List[ProxyType] = []
self._proxy_to_index: Dict[str, int] = {} # O(1) lookup by unique key (server + username)
for i, proxy in enumerate(proxies):
if isinstance(proxy, (str, dict)):
if isinstance(proxy, dict) and "server" not in proxy:
raise ValueError("Proxy dict must have a 'server' key")
self._proxy_to_index[_get_proxy_key(proxy)] = i
self._proxies.append(proxy)
else:
raise TypeError(f"Invalid proxy type: {type(proxy)}. Expected str or dict.")
self._current_index = 0
|
__slots__ = (
"_proxies",
"_proxy_to_index",
"_strategy",
"_current_index",
"_lock",
)
Get a copy of all configured proxies.
Get the next proxy according to the rotation strategy.
Source code in scrapling/engines/toolbelt/proxy_rotation.py
| def get_proxy(self) -> ProxyType:
"""Get the next proxy according to the rotation strategy."""
with self._lock:
proxy, self._current_index = self._strategy(self._proxies, self._current_index)
return proxy
|
Return the total number of configured proxies.
Source code in scrapling/engines/toolbelt/proxy_rotation.py
| def __len__(self) -> int:
"""Return the total number of configured proxies."""
return len(self._proxies)
|
Source code in scrapling/engines/toolbelt/proxy_rotation.py
| def __repr__(self) -> str:
return f"ProxyRotator(proxies={len(self._proxies)})"
|