feat: add TemporarilyUnavailableError for 503 responses

Author: boutetnico

src/rdapapi/__init__.py

@@ -8,6 +8,7 @@
     RateLimitError,
     RdapApiError,
     SubscriptionRequiredError,
+    TemporarilyUnavailableError,
     UpstreamError,
     ValidationError,
 )
@@ -43,6 +44,7 @@
     "NotFoundError",
     "ValidationError",
     "RateLimitError",
+    "TemporarilyUnavailableError",
     "UpstreamError",
     # Models
     "AsnResponse",

src/rdapapi/client.py

@@ -13,6 +13,7 @@
     RateLimitError,
     RdapApiError,
     SubscriptionRequiredError,
+    TemporarilyUnavailableError,
     UpstreamError,
     ValidationError,
 )
@@ -36,6 +37,7 @@
     404: NotFoundError,
     429: RateLimitError,
     502: UpstreamError,
+    503: TemporarilyUnavailableError,
 }
 
 
@@ -56,7 +58,7 @@ def _raise_for_status(response: httpx.Response) -> None:
             "error": error,
         }
 
-        if exc_class is RateLimitError:
+        if exc_class in (RateLimitError, TemporarilyUnavailableError):
             retry_after = response.headers.get("Retry-After")
             kwargs["retry_after"] = int(retry_after) if retry_after else None
 

src/rdapapi/exceptions.py

@@ -9,6 +9,7 @@
     "NotFoundError",
     "ValidationError",
     "RateLimitError",
+    "TemporarilyUnavailableError",
     "UpstreamError",
 ]
 
@@ -60,5 +61,20 @@ def __init__(
         self.retry_after = retry_after
 
 
+class TemporarilyUnavailableError(RdapApiError):
+    """Raised when the domain data is temporarily unavailable (HTTP 503)."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        status_code: int | None = None,
+        error: str | None = None,
+        retry_after: int | None = None,
+    ) -> None:
+        super().__init__(message, status_code=status_code, error=error)
+        self.retry_after = retry_after
+
+
 class UpstreamError(RdapApiError):
     """Raised when the upstream RDAP server fails (HTTP 502)."""

tests/test_async.py

@@ -4,7 +4,7 @@
 import pytest
 import respx
 
-from rdapapi import AsyncRdapApi, AuthenticationError, NotFoundError, RateLimitError, SubscriptionRequiredError
+from rdapapi import AsyncRdapApi, AuthenticationError, NotFoundError, RateLimitError, SubscriptionRequiredError, TemporarilyUnavailableError
 from rdapapi.models import BulkDomainResponse
 
 BASE_URL = "https://rdapapi.io/api/v1"
@@ -198,6 +198,24 @@ async def test_async_rate_limit_error():
     assert exc_info.value.retry_after == 30
 
 
+@pytest.mark.asyncio
+@respx.mock
+async def test_async_temporarily_unavailable_error():
+    respx.get(f"{BASE_URL}/domain/test.com").mock(
+        return_value=httpx.Response(
+            503,
+            json={"error": "temporarily_unavailable", "message": "Data for this domain is temporarily unavailable."},
+            headers={"Retry-After": "300"},
+        )
+    )
+
+    async with AsyncRdapApi("test-key", base_url=BASE_URL) as api:
+        with pytest.raises(TemporarilyUnavailableError) as exc_info:
+            await api.domain("test.com")
+
+    assert exc_info.value.retry_after == 300
+
+
 # === Async Bulk Domain Lookups ===
 
 BULK_RESPONSE = {

tests/test_client.py

@@ -11,6 +11,7 @@
     RdapApi,
     RdapApiError,
     SubscriptionRequiredError,
+    TemporarilyUnavailableError,
     UpstreamError,
     ValidationError,
 )
@@ -333,6 +334,43 @@ def test_rate_limit_error():
     api.close()
 
 
+@respx.mock
+def test_temporarily_unavailable_error():
+    respx.get(f"{BASE_URL}/domain/test.com").mock(
+        return_value=httpx.Response(
+            503,
+            json={"error": "temporarily_unavailable", "message": "Data for this domain is temporarily unavailable."},
+            headers={"Retry-After": "300"},
+        )
+    )
+
+    api = RdapApi("test-key", base_url=BASE_URL)
+    with pytest.raises(TemporarilyUnavailableError) as exc_info:
+        api.domain("test.com")
+
+    assert exc_info.value.status_code == 503
+    assert exc_info.value.retry_after == 300
+    api.close()
+
+
+@respx.mock
+def test_temporarily_unavailable_error_without_retry_after():
+    respx.get(f"{BASE_URL}/domain/test.com").mock(
+        return_value=httpx.Response(
+            503,
+            json={"error": "temporarily_unavailable", "message": "Data for this domain is temporarily unavailable."},
+        )
+    )
+
+    api = RdapApi("test-key", base_url=BASE_URL)
+    with pytest.raises(TemporarilyUnavailableError) as exc_info:
+        api.domain("test.com")
+
+    assert exc_info.value.status_code == 503
+    assert exc_info.value.retry_after is None
+    api.close()
+
+
 @respx.mock
 def test_validation_error():
     respx.get(f"{BASE_URL}/domain/invalid").mock(