Validate if proxy IP is valid :param proxy: :return:
(self, proxy: IpInfoModel)
| 69 | self.proxy_list = await self.ip_provider.get_proxy(self.ip_pool_count) |
| 70 | |
| 71 | async def _is_valid_proxy(self, proxy: IpInfoModel) -> bool: |
| 72 | """ |
| 73 | Validate if proxy IP is valid |
| 74 | :param proxy: |
| 75 | :return: |
| 76 | """ |
| 77 | utils.logger.info( |
| 78 | f"[ProxyIpPool._is_valid_proxy] testing {proxy.ip} is it valid " |
| 79 | ) |
| 80 | try: |
| 81 | # httpx 0.28.1 requires passing proxy URL string directly, not a dictionary |
| 82 | if proxy.user and proxy.password: |
| 83 | proxy_url = f"http://{proxy.user}:{proxy.password}@{proxy.ip}:{proxy.port}" |
| 84 | else: |
| 85 | proxy_url = f"http://{proxy.ip}:{proxy.port}" |
| 86 | |
| 87 | async with make_async_client(proxy=proxy_url) as client: |
| 88 | response = await client.get(self.valid_ip_url) |
| 89 | if response.status_code == 200: |
| 90 | return True |
| 91 | else: |
| 92 | return False |
| 93 | except Exception as e: |
| 94 | utils.logger.info( |
| 95 | f"[ProxyIpPool._is_valid_proxy] testing {proxy.ip} err: {e}" |
| 96 | ) |
| 97 | raise e |
| 98 | |
| 99 | @retry(stop=stop_after_attempt(3), wait=wait_fixed(1)) |
| 100 | async def get_proxy(self) -> IpInfoModel: |
no test coverage detected