From 7e24f77bde22af18f4d4b7ad043757da328715a9 Mon Sep 17 00:00:00 2001 From: Silvano Cerza Date: Wed, 21 Jan 2026 15:50:24 +0100 Subject: [PATCH] Add resproxy API support --- ipinfo/handler.py | 52 +++++++++++++++++++++++++++++++++++++ ipinfo/handler_async.py | 51 ++++++++++++++++++++++++++++++++++++ ipinfo/handler_utils.py | 3 +++ tests/handler_async_test.py | 35 +++++++++++++++++++++++++ tests/handler_test.py | 31 ++++++++++++++++++++++ 5 files changed, 172 insertions(+) diff --git a/ipinfo/handler.py b/ipinfo/handler.py index 507c810..65bf88a 100644 --- a/ipinfo/handler.py +++ b/ipinfo/handler.py @@ -13,6 +13,7 @@ from .exceptions import RequestQuotaExceededError, TimeoutExceededError from .handler_utils import ( API_URL, + RESPROXY_API_URL, BATCH_MAX_SIZE, CACHE_MAXSIZE, CACHE_TTL, @@ -145,6 +146,57 @@ def getDetails(self, ip_address=None, timeout=None): return Details(details) + def getResproxy(self, ip_address, timeout=None): + """ + Get residential proxy information for specified IP address. + + Returns a Details object containing: + - ip: The IP address + - last_seen: The last recorded date when the proxy was active (YYYY-MM-DD) + - percent_days_seen: Percentage of days active in the last 7-day period + - service: Name of the residential proxy service + + If `timeout` is not `None`, it will override the client-level timeout + just for this operation. + """ + if isinstance(ip_address, IPv4Address) or isinstance(ip_address, IPv6Address): + ip_address = ip_address.exploded + + # check cache first. + cache_key_str = f"resproxy:{ip_address}" + try: + cached_data = self.cache[cache_key(cache_key_str)] + return Details(cached_data) + except KeyError: + pass + + # prepare req http opts + req_opts = {**self.request_options} + if timeout is not None: + req_opts["timeout"] = timeout + + # do http req + url = f"{RESPROXY_API_URL}/{ip_address}" + headers = handler_utils.get_headers(self.access_token, self.headers) + response = requests.get(url, headers=headers, **req_opts) + if response.status_code == 429: + raise RequestQuotaExceededError() + if response.status_code >= 400: + error_code = response.status_code + content_type = response.headers.get("Content-Type") + if content_type == "application/json": + error_response = response.json() + else: + error_response = {"error": response.text} + raise APIError(error_code, error_response) + details = response.json() + + # cache result + self.cache[cache_key(cache_key_str)] = details + + return Details(details) + + def getBatchDetails( self, ip_addresses, diff --git a/ipinfo/handler_async.py b/ipinfo/handler_async.py index c71357a..af6a486 100644 --- a/ipinfo/handler_async.py +++ b/ipinfo/handler_async.py @@ -15,6 +15,7 @@ from .exceptions import RequestQuotaExceededError, TimeoutExceededError from .handler_utils import ( API_URL, + RESPROXY_API_URL, BATCH_MAX_SIZE, CACHE_MAXSIZE, CACHE_TTL, @@ -167,6 +168,56 @@ async def getDetails(self, ip_address=None, timeout=None): return Details(details) + async def getResproxy(self, ip_address, timeout=None): + """ + Get residential proxy information for specified IP address. + + Returns a Details object containing: + - ip: The IP address + - last_seen: The last recorded date when the proxy was active (YYYY-MM-DD) + - percent_days_seen: Percentage of days active in the last 7-day period + - service: Name of the residential proxy service + + If `timeout` is not `None`, it will override the client-level timeout + just for this operation. + """ + self._ensure_aiohttp_ready() + + if isinstance(ip_address, IPv4Address) or isinstance(ip_address, IPv6Address): + ip_address = ip_address.exploded + + # check cache first. + cache_key_str = f"resproxy:{ip_address}" + try: + cached_data = self.cache[cache_key(cache_key_str)] + return Details(cached_data) + except KeyError: + pass + + # do http req + url = f"{RESPROXY_API_URL}/{ip_address}" + headers = handler_utils.get_headers(self.access_token, self.headers) + req_opts = {} + if timeout is not None: + req_opts["timeout"] = timeout + async with self.httpsess.get(url, headers=headers, **req_opts) as resp: + if resp.status == 429: + raise RequestQuotaExceededError() + if resp.status >= 400: + error_code = resp.status + content_type = resp.headers.get("Content-Type") + if content_type == "application/json": + error_response = await resp.json() + else: + error_response = {"error": resp.text()} + raise APIError(error_code, error_response) + details = await resp.json() + + # cache result + self.cache[cache_key(cache_key_str)] = details + + return Details(details) + async def getBatchDetails( self, ip_addresses, diff --git a/ipinfo/handler_utils.py b/ipinfo/handler_utils.py index 9beb833..f37bbfb 100644 --- a/ipinfo/handler_utils.py +++ b/ipinfo/handler_utils.py @@ -21,6 +21,9 @@ # Base URL for the IPinfo Plus API (same as Core) PLUS_API_URL = "https://api.ipinfo.io/lookup" +# Base URL for the IPinfo Residential Proxy API +RESPROXY_API_URL = "https://ipinfo.io/resproxy" + # Base URL to get country flag image link. # "PK" -> "https://cdn.ipinfo.io/static/images/countries-flags/PK.svg" COUNTRY_FLAGS_URL = "https://cdn.ipinfo.io/static/images/countries-flags/" diff --git a/tests/handler_async_test.py b/tests/handler_async_test.py index 6cc1011..2520932 100644 --- a/tests/handler_async_test.py +++ b/tests/handler_async_test.py @@ -252,3 +252,38 @@ async def test_bogon_details(): handler = AsyncHandler(token) details = await handler.getDetails("127.0.0.1") assert details.all == {"bogon": True, "ip": "127.0.0.1"} + + +################# +# RESPROXY TESTS +################# + + +@pytest.mark.asyncio +async def test_get_resproxy(): + token = os.environ.get("IPINFO_TOKEN", "") + if not token: + pytest.skip("token required for resproxy tests") + handler = AsyncHandler(token) + # Use an IP known to be a residential proxy (from API documentation) + details = await handler.getResproxy("175.107.211.204") + assert isinstance(details, Details) + assert details.ip == "175.107.211.204" + assert details.last_seen is not None + assert details.percent_days_seen is not None + assert details.service is not None + await handler.deinit() + + +@pytest.mark.asyncio +async def test_get_resproxy_caching(): + token = os.environ.get("IPINFO_TOKEN", "") + if not token: + pytest.skip("token required for resproxy tests") + handler = AsyncHandler(token) + # First call should hit the API + details1 = await handler.getResproxy("175.107.211.204") + # Second call should hit the cache + details2 = await handler.getResproxy("175.107.211.204") + assert details1.ip == details2.ip + await handler.deinit() \ No newline at end of file diff --git a/tests/handler_test.py b/tests/handler_test.py index 329753d..c16cdbe 100644 --- a/tests/handler_test.py +++ b/tests/handler_test.py @@ -236,3 +236,34 @@ def test_iterative_bogon_details(): handler = Handler(token) details = next(handler.getBatchDetailsIter(["127.0.0.1"])) assert details.all == {"bogon": True, "ip": "127.0.0.1"} + + +################# +# RESPROXY TESTS +################# + + +def test_get_resproxy(): + token = os.environ.get("IPINFO_TOKEN", "") + if not token: + pytest.skip("token required for resproxy tests") + handler = Handler(token) + # Use an IP known to be a residential proxy (from API documentation) + details = handler.getResproxy("175.107.211.204") + assert isinstance(details, Details) + assert details.ip == "175.107.211.204" + assert details.last_seen is not None + assert details.percent_days_seen is not None + assert details.service is not None + + +def test_get_resproxy_caching(): + token = os.environ.get("IPINFO_TOKEN", "") + if not token: + pytest.skip("token required for resproxy tests") + handler = Handler(token) + # First call should hit the API + details1 = handler.getResproxy("175.107.211.204") + # Second call should hit the cache + details2 = handler.getResproxy("175.107.211.204") + assert details1.ip == details2.ip \ No newline at end of file