Proxy-Rotator, der sicher aus mehreren Threads genutzt werden kann und austauschbare Wechselstrategien bietet.
Unterstützt zyklischen Wechsel als Standard, eigene Strategien über aufrufbare Objekte sowie Proxys als URL-Zeichenketten oder Dictionarys im Playwright-Format.
Initialisiert den Proxy-Rotator.
| PARAMETER |
BESCHREIBUNG |
proxies
|
Liste von Proxy-URLs oder Proxy-Dictionarys im Playwright-Format. Zeichenkettenformat: "http://proxy1:8080" oder "http://user:pass@proxy:8080". Dictionary-Format: {"server": "http://proxy:8080", "username": "user", "password": "pass"}.
TYP:
List[ProxyType]
|
strategy
|
Funktion für die Wechselstrategie. Nimmt (proxies, current_index) an und gibt (proxy, next_index) zurück. Der Standardwert ist cyclic_rotation.
TYP:
RotationStrategy
STANDARDWERT:
cyclic_rotation
|
Quellcode 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",
)
Liefert eine Kopie aller konfigurierten Proxys.
Liefert den nächsten Proxy nach der gewählten Wechselstrategie.
Quellcode 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
|
Gibt die Gesamtzahl der konfigurierten Proxys zurück.
Quellcode in scrapling/engines/toolbelt/proxy_rotation.py
| def __len__(self) -> int:
"""Return the total number of configured proxies."""
return len(self._proxies)
|
Quellcode in scrapling/engines/toolbelt/proxy_rotation.py
| def __repr__(self) -> str:
return f"ProxyRotator(proxies={len(self._proxies)})"
|