docs: fix README health-check snippet's invalid aiohttp kwarg; widen add()/replace() type hints to accept dict shapes

the flagship health-check snippet called session.get(url, proxies=proxy) on a
plain aiohttp session; aiohttp's ClientSession.get/_request has no `proxies=`
kwarg (only singular `proxy:`), so the snippet TypeErrors immediately if
followed verbatim. pm.get()'s dict output is documented (aiohttp()'s own
docstring) as being for aioweb's ExtendedSession(proxies=...), so the snippet
now names that session type instead of a bare aiohttp one.

add()/replace() type hints omitted Dict[str, str] even though both already
call to_proxy() at runtime and accept dict-shaped proxy specs, same as the
burn()/restore()/is_burned()/remove() family. purely annotation, no behavior
change.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
2026-07-06 00:15:29 -04:00
parent a4a0227d5d
commit 1adc8139b9
2 changed files with 5 additions and 3 deletions
+3 -1
View File
@@ -105,12 +105,14 @@ regardless of source.)
```python
from aioproxies import AioProxies, ProxiesExhaustedError
from aioweb import ExtendedSession
pm = AioProxies(proxies=[...], cooldown=5) # 5s reuse spacing; cooldown defaults to 0 (off)
try:
proxy = pm.get() # next usable proxy, aiohttp dict
resp = await session.get(url, proxies=proxy)
async with ExtendedSession(proxies=proxy) as session:
resp = await session.get(url)
if response_looks_blocked(resp):
pm.burn(proxy, 600) # time out 10 min ... or pm.burn(proxy) for dead
except ProxiesExhaustedError:
+2 -2
View File
@@ -302,7 +302,7 @@ class AioProxies:
for state in self._state.values():
state["uses"] = 0
def replace(self, proxies: List[Union[str, Proxy]], *, keep_state: bool = False) -> None:
def replace(self, proxies: List[Union[str, Proxy, Dict[str, str]]], *, keep_state: bool = False) -> None:
"""swap the entire proxy list
keep_state=False (default) wipes all per-proxy state (fresh batch).
@@ -336,7 +336,7 @@ class AioProxies:
self._state = new_state
self._index = 0
def add(self, proxies: Union[str, Proxy, List[Union[str, Proxy]]]) -> None:
def add(self, proxies: Union[str, Proxy, Dict[str, str], List[Union[str, Proxy, Dict[str, str]]]]) -> None:
"""append proxies to the pool, keeping existing state; skip duplicate keys"""
if not self._is_list_source():
self._warn_non_list("add()")