feat: add bulk domain lookups

Author: boutetnico

README.md

@@ -46,6 +46,24 @@ print(entity.autnums[0].handle) # "AS15169"
 api.close()
 ```
 
+## Bulk domain lookups
+
+Look up multiple domains in a single request (Pro and Business plans). Up to 10 domains per call, with concurrent upstream fetches:
+
+```python
+result = api.bulk_domains(["google.com", "github.com", "invalid..com"], follow=True)
+
+print(result.summary)  # total=3, successful=2, failed=1
+
+for r in result.results:
+    if r.status == "success":
+        print(f"{r.data.domain}: {r.data.registrar.name}")
+    else:
+        print(f"{r.domain}: {r.error}")
+```
+
+Each domain counts as one request toward your monthly quota. Starter plans receive a `SubscriptionRequiredError` (403).
+
 ## Registrar follow-through
 
 For thin registries like `.com` and `.net`, the registry only returns basic registrar info. Use `follow=True` to follow the registrar's RDAP link and get richer contact data:

examples/bulk_lookup.py

@@ -1,16 +1,21 @@
-"""Look up multiple domains in a loop."""
+"""Look up multiple domains in a single bulk request (Pro/Business plans)."""
 
-from rdapapi import NotFoundError, RdapApi
+from rdapapi import RdapApi
 
 api = RdapApi("your-api-key")
 
-domains = ["google.com", "github.com", "cloudflare.com", "nonexistent.example"]
+# Bulk lookup — up to 10 domains in one request
+result = api.bulk_domains(
+    ["google.com", "github.com", "invalid..com"],
+    follow=True,
+)
 
-for name in domains:
-    try:
-        result = api.domain(name)
-        print(f"{result.domain}: registrar={result.registrar.name}, expires={result.dates.expires}")
-    except NotFoundError:
-        print(f"{name}: not found")
+print(f"Total: {result.summary.total}, OK: {result.summary.successful}, Failed: {result.summary.failed}")
+
+for r in result.results:
+    if r.status == "success":
+        print(f"  {r.data.domain}: registrar={r.data.registrar.name}, expires={r.data.dates.expires}")
+    else:
+        print(f"  {r.domain}: {r.error} — {r.message}")
 
 api.close()

src/rdapapi/__init__.py

@@ -13,6 +13,9 @@
 )
 from .models import (
     AsnResponse,
+    BulkDomainResponse,
+    BulkDomainResult,
+    BulkDomainSummary,
     Contact,
     Dates,
     DomainResponse,
@@ -43,6 +46,9 @@
     "UpstreamError",
     # Models
     "AsnResponse",
+    "BulkDomainResponse",
+    "BulkDomainResult",
+    "BulkDomainSummary",
     "Contact",
     "Dates",
     "DomainResponse",

src/rdapapi/client.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 import httpx
 
@@ -18,6 +18,7 @@
 )
 from .models import (
     AsnResponse,
+    BulkDomainResponse,
     DomainResponse,
     EntityResponse,
     IpResponse,
@@ -62,6 +63,14 @@ def _raise_for_status(response: httpx.Response) -> None:
         raise exc_class(message, **kwargs)
 
 
+def _parse_bulk_response(data: dict) -> BulkDomainResponse:
+    """Parse a bulk domain response, merging meta into each successful result's data."""
+    for result in data.get("results", []):
+        if result.get("status") == "success" and "data" in result and "meta" in result:
+            result["data"]["meta"] = result.pop("meta")
+    return BulkDomainResponse.model_validate(data)
+
+
 class RdapApi:
     """Synchronous client for the RDAP API.
 
@@ -102,6 +111,15 @@ def _request(
         _raise_for_status(response)
         return response.json()
 
+    def _post(
+        self,
+        path: str,
+        body: Dict[str, Any],
+    ) -> dict:
+        response = self._client.post(path, json=body)
+        _raise_for_status(response)
+        return response.json()
+
     def domain(self, name: str, *, follow: bool = False) -> DomainResponse:
         """Look up RDAP registration data for a domain name."""
         params = {"follow": "true"} if follow else None
@@ -132,6 +150,23 @@ def entity(self, handle: str) -> EntityResponse:
         data = self._request(f"/entity/{handle}")
         return EntityResponse.model_validate(data)
 
+    def bulk_domains(
+        self,
+        domains: List[str],
+        *,
+        follow: bool = False,
+    ) -> BulkDomainResponse:
+        """Look up multiple domains in a single request.
+
+        Sends up to 10 domains concurrently. Requires a Pro or Business plan.
+        Each domain counts as one request toward your quota.
+        """
+        body: Dict[str, Any] = {"domains": domains}
+        if follow:
+            body["follow"] = True
+        data = self._post("/domains/bulk", body)
+        return _parse_bulk_response(data)
+
     def close(self) -> None:
         """Close the underlying HTTP client."""
         self._client.close()
@@ -183,6 +218,15 @@ async def _request(
         _raise_for_status(response)
         return response.json()
 
+    async def _post(
+        self,
+        path: str,
+        body: Dict[str, Any],
+    ) -> dict:
+        response = await self._client.post(path, json=body)
+        _raise_for_status(response)
+        return response.json()
+
     async def domain(self, name: str, *, follow: bool = False) -> DomainResponse:
         """Look up RDAP registration data for a domain name."""
         params = {"follow": "true"} if follow else None
@@ -213,6 +257,23 @@ async def entity(self, handle: str) -> EntityResponse:
         data = await self._request(f"/entity/{handle}")
         return EntityResponse.model_validate(data)
 
+    async def bulk_domains(
+        self,
+        domains: List[str],
+        *,
+        follow: bool = False,
+    ) -> BulkDomainResponse:
+        """Look up multiple domains in a single request.
+
+        Sends up to 10 domains concurrently. Requires a Pro or Business plan.
+        Each domain counts as one request toward your quota.
+        """
+        body: Dict[str, Any] = {"domains": domains}
+        if follow:
+            body["follow"] = True
+        data = await self._post("/domains/bulk", body)
+        return _parse_bulk_response(data)
+
     async def close(self) -> None:
         """Close the underlying HTTP client."""
         await self._client.aclose()

src/rdapapi/models.py

@@ -8,6 +8,9 @@
 
 __all__ = [
     "AsnResponse",
+    "BulkDomainResponse",
+    "BulkDomainResult",
+    "BulkDomainSummary",
     "Contact",
     "Dates",
     "DomainResponse",
@@ -183,6 +186,36 @@ class EntityNetwork(BaseModel):
     cidr: List[str] = Field(default_factory=list)
 
 
+class BulkDomainResult(BaseModel):
+    """A single result within a bulk domain lookup response.
+
+    When ``status`` is ``"success"``, ``data`` contains a full
+    :class:`DomainResponse`.  When ``status`` is ``"error"``,
+    ``error`` and ``message`` describe the failure.
+    """
+
+    domain: str
+    status: str
+    data: Optional[DomainResponse] = None
+    error: Optional[str] = None
+    message: Optional[str] = None
+
+
+class BulkDomainSummary(BaseModel):
+    """Summary counts for a bulk domain lookup."""
+
+    total: int
+    successful: int
+    failed: int
+
+
+class BulkDomainResponse(BaseModel):
+    """Response from a bulk domain lookup."""
+
+    results: List[BulkDomainResult]
+    summary: BulkDomainSummary
+
+
 class EntityResponse(BaseModel):
     """Response from an entity lookup."""
 

tests/test_async.py

@@ -4,7 +4,8 @@
 import pytest
 import respx
 
-from rdapapi import AsyncRdapApi, AuthenticationError, NotFoundError, RateLimitError
+from rdapapi import AsyncRdapApi, AuthenticationError, NotFoundError, RateLimitError, SubscriptionRequiredError
+from rdapapi.models import BulkDomainResponse
 
 BASE_URL = "https://rdapapi.io/api/v1"
 
@@ -123,3 +124,69 @@ async def test_async_rate_limit_error():
             await api.domain("test.com")
 
     assert exc_info.value.retry_after == 30
+
+
+# === Async Bulk Domain Lookups ===
+
+BULK_RESPONSE = {
+    "results": [
+        {
+            "domain": "google.com",
+            "status": "success",
+            "data": {
+                "domain": "google.com",
+                "unicode_name": None,
+                "handle": "2138514_DOMAIN_COM-VRSN",
+                "status": ["client delete prohibited"],
+                "registrar": {
+                    "name": "MarkMonitor Inc.",
+                    "iana_id": "292",
+                    "abuse_email": None,
+                    "abuse_phone": None,
+                    "url": None,
+                },
+                "dates": {"registered": "1997-09-15T04:00:00Z", "expires": "2028-09-14T04:00:00Z", "updated": None},
+                "nameservers": ["ns1.google.com"],
+                "dnssec": False,
+                "entities": {},
+            },
+            "meta": {
+                "rdap_server": "https://rdap.verisign.com/com/v1/",
+                "raw_rdap_url": "https://rdap.verisign.com/com/v1/domain/google.com",
+                "cached": False,
+                "cache_expires": "2026-02-25T15:30:00Z",
+            },
+        },
+    ],
+    "summary": {"total": 1, "successful": 1, "failed": 0},
+}
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_async_bulk_domains_lookup():
+    respx.post(f"{BASE_URL}/domains/bulk").mock(return_value=httpx.Response(200, json=BULK_RESPONSE))
+
+    async with AsyncRdapApi("test-key", base_url=BASE_URL) as api:
+        result = await api.bulk_domains(["google.com"])
+
+    assert isinstance(result, BulkDomainResponse)
+    assert result.summary.successful == 1
+    assert result.results[0].data is not None
+    assert result.results[0].data.domain == "google.com"
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_async_bulk_domains_plan_upgrade_required():
+    respx.post(f"{BASE_URL}/domains/bulk").mock(
+        return_value=httpx.Response(
+            403, json={"error": "plan_upgrade_required", "message": "Bulk lookups require a Pro or Business plan."}
+        )
+    )
+
+    async with AsyncRdapApi("test-key", base_url=BASE_URL) as api:
+        with pytest.raises(SubscriptionRequiredError) as exc_info:
+            await api.bulk_domains(["google.com"])
+
+    assert exc_info.value.error == "plan_upgrade_required"