Randomly extract a proxy IP from the proxy pool :return:
(self)
| 98 | |
| 99 | @retry(stop=stop_after_attempt(3), wait=wait_fixed(1)) |
| 100 | async def get_proxy(self) -> IpInfoModel: |
| 101 | """ |
| 102 | Randomly extract a proxy IP from the proxy pool |
| 103 | :return: |
| 104 | """ |
| 105 | if len(self.proxy_list) == 0: |
| 106 | await self._reload_proxies() |
| 107 | |
| 108 | proxy = random.choice(self.proxy_list) |
| 109 | self.proxy_list.remove(proxy) # Remove an IP once extracted |
| 110 | if self.enable_validate_ip: |
| 111 | if not await self._is_valid_proxy(proxy): |
| 112 | raise Exception( |
| 113 | "[ProxyIpPool.get_proxy] current ip invalid and again get it" |
| 114 | ) |
| 115 | self.current_proxy = proxy # Save currently used proxy |
| 116 | return proxy |
| 117 | |
| 118 | def is_current_proxy_expired(self, buffer_seconds: int = 30) -> bool: |
| 119 | """ |
no test coverage detected