Designing a data enrichment pipeline with caching and bounded concurrency

# ======================================================================
# Q: How do you design an enrichment pipeline (e.g. adding geo-IP, owner, CVE to
# each asset)?
#
# Stages with a clear responsibility: ingest → validate → normalize → enrich →
# store. Each stage is idempotent and records its state (status, attempts,
# last_error), so that only what failed is replayed. Enrichment from external
# APIs is done in batches, with a cache (I don't ask about the same IP twice), a
# rate limit and a circuit breaker.
# ======================================================================

import asyncio, httpx

async def enrich_ips(ips: list[str], concurrency: int = 20) -> dict[str, dict]:
    sem = asyncio.Semaphore(concurrency)         # limit on concurrent calls
    async with httpx.AsyncClient(timeout=5) as client:
        async def one(ip):
            if (hit := await cache.get(f'geo:{ip}')) is not None:
                return ip, hit
            async with sem:
                r = await client.get(f'{GEO_URL}/{ip}')
                r.raise_for_status()
                data = r.json()
            await cache.set(f'geo:{ip}', data, ex=86400)
            return ip, data
        results = await asyncio.gather(*(one(ip) for ip in set(ips)),
                                       return_exceptions=True)
    return {r[0]: r[1] for r in results if not isinstance(r, Exception)}