Building a Python Domain Lookup Tool with Redis Caching and Rate Limits

A domain lookup feels simple until someone clicks the search button six times. Your application sends six identical requests. The upstream service slows down. Your workers wait. A useful feature becomes the reason your application feels stuck. A small Redis cache helps, but a dependable service also needs clear time limits, sensible request budgets, and honest error messages.

Give Each Request a Predictable Path

We will use FastAPI for the HTTP endpoint, HTTPX for outgoing requests, and Redis for cached responses and shared traffic controls. The application will expose GET /domains/{domain}. A successful response will contain the normalized domain, the provider data, and a flag showing whether Redis supplied the answer.

Keep the first version small. Accept a domain name rather than a complete website address. Reject paths, ports, email addresses, and IP addresses. This makes cache keys predictable and prevents callers from turning your endpoint into a general network request tool.

The service will follow these rules:

• Normalize and validate input before accessing Redis or making network requests.
• Cache successful registration responses for a short, configurable period.
• Apply one shared budget to outgoing requests across application workers.
• Respect upstream throttling and return temporary failures without inventing registration facts.

The outgoing budget protects the provider. It does not replace user authentication or a separate incoming limit. Before exposing this service publicly, apply a per-account request limit at your gateway. Otherwise, one caller can consume the shared budget by submitting many different domains.

Prepare Python and Redis

Use Python 3.11 or newer. Create a virtual environment and install the application dependencies. Keep Redis on a private interface. For local development, an existing Redis instance on your machine is enough.

python -m venv .venv
source .venv/bin/activate
pip install fastapi "uvicorn[standard]" httpx redis idna

Put the following code in app.py. The snippets in this article belong in that file, in the order shown. Configuration comes from environment variables so the same application can run locally or inside a container.

import asyncio
import ipaddress
import json
import os
import re
import time
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime

import httpx
import idna
from fastapi import FastAPI, HTTPException
from redis.asyncio import Redis
from redis.exceptions import RedisError

CACHE_TTL = int(os.getenv("CACHE_TTL_SECONDS", "300"))
OUTBOUND_LIMIT = int(os.getenv("OUTBOUND_REQUESTS_PER_MINUTE", "10"))
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
LOOKUP_BASE = os.environ["LOOKUP_BASE_URL"].rstrip("/")

if CACHE_TTL < 1 or OUTBOUND_LIMIT < 1:
    raise ValueError("Cache lifetime and request limit must be positive")


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = Redis.from_url(
        REDIS_URL,
        decode_responses=True,
        socket_connect_timeout=1,
        socket_timeout=1,
    )
    app.state.http = httpx.AsyncClient(
        timeout=httpx.Timeout(
            connect=2.0, read=5.0, write=2.0, pool=1.0
        ),
        limits=httpx.Limits(
            max_connections=20,
            max_keepalive_connections=10,
        ),
        follow_redirects=False,
    )
    try:
        yield
    finally:
        await app.state.http.aclose()
        await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)


def normalize_domain(value: str) -> str:
    value = value.strip().rstrip(".")
    try:
        domain = idna.encode(
            value, uts46=True, std3_rules=True
        ).decode("ascii").lower()
    except idna.IDNAError:
        raise HTTPException(400, "Enter a valid domain name")

    try:
        ipaddress.ip_address(domain)
    except ValueError:
        pass
    else:
        raise HTTPException(400, "Enter a domain, not an IP address")

    labels = domain.split(".")
    if len(domain) > 253 or len(labels) < 2:
        raise HTTPException(400, "Enter a complete domain name")

    for label in labels:
        if not re.fullmatch(
            r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", label
        ):
            raise HTTPException(400, "Enter a valid domain name")

    return domain

Normalization turns different spellings of the same input into one cache key. Uppercase letters become lowercase. A trailing dot disappears. Internationalized names become an ASCII representation through IDNA processing.

This validation checks syntax. It does not prove that a domain exists or accepts registrations. That distinction matters when you write the response text shown to users.

Connect the Provider Without Guessing Its Fields

At this point, configure the upstream adapter. Consult lookup API documentation for the GET request path and supported response content. Set LOOKUP_BASE_URL to the documented origin followed by /api/v1/lookup/rdap. The application appends the validated domain. The documented public endpoint currently requires no API key.

The documented response envelope contains tool, data, and methodology. Map these into provider_tool, registration, and methodology. Preserve the data object intact because the reference does not define a complete property-by-property schema for its RDAP summary, events, nameservers, notices, and raw JSON.

This endpoint is intended for occasional interactive use, without a bulk-throughput or uptime promise. Keep this tutorial deployment modest. Production-scale integrations should use an appropriate authoritative source or contracted provider. Cached observations also need their original timestamps and source references.

Our own application schema stays deliberately small. If you later add a dedicated expiration field, map it from a verified response fixture. Match the event action explicitly, retain its date, and allow a missing value. Do not assume the first event means expiration.

Share the Outgoing Budget Through Redis

A counter stored inside Python would apply separately to each worker. Redis gives the workers one shared budget. The script below increments a counter and sets its expiry atomically. That prevents a process interruption between those operations from leaving a permanent counter behind.

This is a fixed window that starts with the first outgoing attempt. It is easy to understand, although requests can cluster around a window boundary. Choose a conservative budget. A stricter deployment can replace it with a token bucket.

RATE_SCRIPT = """
local count = redis.call('INCR', KEYS[1])
if count == 1 then
    redis.call('EXPIRE', KEYS[1], ARGV[1])
end
return {count, redis.call('TTL', KEYS[1])}
"""

COOLDOWN_SCRIPT = """
local current = redis.call('TTL', KEYS[1])
local requested = tonumber(ARGV[1])
if current < requested then
    redis.call('SET', KEYS[1], '1', 'EX', requested)
end
return 1
"""


async def reserve_outgoing(redis: Redis) -> None:
    cooldown = await redis.ttl("domain:upstream:cooldown")
    if cooldown > 0:
        raise HTTPException(
            503,
            "Registration provider is temporarily paused",
            headers={"Retry-After": str(cooldown)},
        )

    count, remaining = await redis.eval(
        RATE_SCRIPT, 1, "domain:upstream:budget", 60
    )
    if count > OUTBOUND_LIMIT:
        raise HTTPException(
            503,
            "Shared lookup capacity is temporarily exhausted",
            headers={"Retry-After": str(max(1, remaining))},
        )


def retry_delay(value: str | None) -> int:
    if value:
        try:
            seconds = int(value)
        except ValueError:
            try:
                retry_at = parsedate_to_datetime(value)
                if retry_at.tzinfo is None:
                    retry_at = retry_at.replace(tzinfo=timezone.utc)
                seconds = int(
                    retry_at.timestamp() - time.time()
                ) + 1
            except (ValueError, TypeError, OverflowError):
                seconds = 60
        return max(1, seconds)
    return 60

The cooldown is shared too. If one worker receives a throttling response, later cache misses see the pause. Cached answers remain usable because the endpoint checks Redis before reserving an outgoing request.

The cooldown update preserves a longer pause that another worker has already stored. Requests already in flight can still finish. This control reduces fresh traffic after throttling; it cannot recall requests already sent.

Fetch Registration Data Within a Firm Deadline

HTTPX timeouts bound separate network phases. An outer asyncio timeout also bounds the overall fetch operation. This prevents a response that keeps delivering small pieces from occupying a worker indefinitely.

We will not automatically retry inside the request handler. Retries consume the same limited upstream capacity and can multiply traffic during an outage. A caller can try again after the returned waiting period.

async def fetch_registration(
    domain: str, redis: Redis, client: httpx.AsyncClient
) -> dict:
    await reserve_outgoing(redis)

    try:
        async with asyncio.timeout(8):
            response = await client.get(
                f"{LOOKUP_BASE}/{domain}",
                headers={"Accept": "application/json"},
            )
    except (TimeoutError, httpx.TimeoutException):
        raise HTTPException(504, "Registration lookup timed out")
    except httpx.RequestError:
        raise HTTPException(502, "Registration provider could not be reached")

    if response.status_code == 429:
        delay = retry_delay(response.headers.get("Retry-After"))
        await redis.eval(
            COOLDOWN_SCRIPT, 1, "domain:upstream:cooldown", delay
        )
        raise HTTPException(
            503,
            "Registration provider requested a pause",
            headers={"Retry-After": str(delay)},
        )

    if response.status_code in (400, 422):
        raise HTTPException(
            502, "Provider could not complete this lookup"
        )

    if response.status_code != 200:
        raise HTTPException(502, "Unexpected registration provider response")

    try:
        payload = response.json()
    except ValueError:
        raise HTTPException(502, "Provider returned invalid JSON")

    if not isinstance(payload, dict):
        raise HTTPException(502, "Provider response has an invalid shape")
    if not isinstance(payload.get("data"), dict):
        raise HTTPException(502, "Provider response is missing lookup data")

    return {
        "domain": domain,
        "registration": payload["data"],
        "provider_tool": payload.get("tool"),
        "methodology": payload.get("methodology"),
        "fetched_at": datetime.now(timezone.utc).isoformat(),
    }

The 429 branch is defensive handling for an HTTP throttling response. It does not assume the provider promises a particular quota or always sends a retry header. The parser accepts both a seconds value and an HTTP date, with a one-minute fallback.

A provider error becomes an application error. It never becomes “domain available.” Registration data may also omit contact details or contain notices that affect interpretation. Keep those qualifications attached to the returned data.

Cache Successful Results and Serve the Endpoint

A five-minute lifetime is a starting policy for this demonstration. Choose your actual lifetime based on the workflow and the upstream caching terms. Registration checks used for a live decision may need a fresh query. A dashboard showing recent observations can often tolerate a short cache.

The cache key includes a version. Change that version when you change the stored response shape. This avoids reading older JSON as though it followed your new application schema.

@app.get("/domains/{domain}")
async def lookup_domain(domain: str):
    normalized = normalize_domain(domain)
    redis = app.state.redis
    key = f"domain:rdap:v1:{normalized}"

    try:
        cached = await redis.get(key)
        if cached is not None:
            try:
                result = json.loads(cached)
                if not isinstance(result, dict):
                    raise ValueError("Invalid cached object")
            except ValueError:
                await redis.delete(key)
            else:
                return {**result, "cached": True}

        result = await fetch_registration(
            normalized, redis, app.state.http
        )
        await redis.set(key, json.dumps(result), ex=CACHE_TTL)
        return {**result, "cached": False}

    except RedisError:
        raise HTTPException(
            503, "Lookup cache and traffic controls are unavailable"
        )

This service stops new lookups when Redis fails. That is intentional because Redis also holds the shared traffic controls. Bypassing it during an outage would remove the protection precisely when the application is already struggling.

The response includes fetched_at, which records when our application received the result. It does not replace a provider observation timestamp. Keep both concepts separate when showing record age in your interface.

After setting LOOKUP_BASE_URL in your environment, start the development server:

uvicorn app:app --reload

Check Failure Paths Before Opening Access

Start with one domain and make the same request twice. The first successful response should contain "cached": false. The next should contain "cached": true, with the same fetch timestamp. After the configured lifetime passes, a fresh lookup should replace it.

Use a mocked HTTPX transport for failure checks. Simulate a timeout, invalid JSON, a missing data object, and a 429 response. Confirm that failed lookups never enter the result cache. Verify that the shared cooldown blocks a different uncached domain too.

Run the budget test against an isolated Redis instance. Submit distinct domains until the configured allowance is exhausted. Confirm that cached domains remain available. Test Redis failure separately and verify that it produces a controlled 503 response.

There is one deliberate limitation in this starter implementation: simultaneous misses for the same domain can each send a request. The shared budget bounds overall traffic, but it does not merge those requests. Before handling substantial concurrency, add request coalescing with a Redis lock, a unique ownership token, bounded waiting, and an ownership-checked release.

Track cache hits, outgoing attempts, timeout counts, and throttle responses. Avoid logging complete registration payloads by default. Operational counters usually answer the first debugging questions without copying contact information into another storage system.

Keep Repeated Lookups Calm and Predictable

A useful domain lookup service returns more than registration data. It tells callers whether the answer came from a cache, preserves when the observation happened, and explains when a fresh lookup cannot finish.

With validated input, short Redis lifetimes, a shared outgoing budget, and bounded network calls, repeated searches become much easier to manage. Start with this small service, verify its failure paths, and adjust its cache policy around how people actually use the answers.