Add resproxy API support

Author: silvanocerza

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,

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,

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/"

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()

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