Approach: keep an in-memory cache with TTL plus an "in-flight" map of key -> asyncio.Future. When a key is missing/expired, the first coroutine creates a Future and kicks off the fetcher (in a background Task). Other concurrent coroutines await the same Future. The fetch task sets result/exception on the Future and always removes the in-flight entry, ensuring waiters never get stuck even on exceptions or cancellations.python
import asyncio
import time
from typing import Any, Dict, Optional
_TTL = 5.0 # seconds; adjustable
class SingleFlightCache:
def __init__(self, ttl: float = _TTL):
self._ttl = ttl
self._cache: Dict[str, (Any, float)] = {} # key -> (value, expiry_ts)
self._inflight: Dict[str, asyncio.Future] = {} # key -> Future
self._lock = asyncio.Lock() # guards _inflight and _cache mutations
async def _fetch(self, key: str) -> Any:
"""
Placeholder for the real fetch. In real use, replace with network/db call.
Simulate I/O with asyncio.sleep.
"""
await asyncio.sleep(0.1)
return f"value-for-{key}"
async def get(self, key: str) -> Any:
# Fast path: check cache without lock (OK for eventual consistency but protect expiry checks)
entry = self._cache.get(key)
now = time.monotonic()
if entry and entry[1] > now:
return entry[0]
# Need to fetch; coordinate via inflight future
async with self._lock:
# Re-check under lock to avoid race
entry = self._cache.get(key)
now = time.monotonic()
if entry and entry[1] > now:
return entry[0]
fut = self._inflight.get(key)
if fut is None:
# create future and spawn fetch task
fut = asyncio.get_event_loop().create_future()
self._inflight[key] = fut
async def _do_fetch():
try:
val = await self._fetch(key)
expiry = time.monotonic() + self._ttl
# set cache under lock to avoid races with other writers
async with self._lock:
self._cache[key] = (val, expiry)
if not fut.done():
fut.set_result(val)
except Exception as exc:
if not fut.done():
fut.set_exception(exc)
finally:
# ensure inflight cleared under lock so new attempts can start
async with self._lock:
self._inflight.pop(key, None)
# run fetch in background so setter is independent of caller cancellation
asyncio.create_task(_do_fetch())
# At this point, fut exists and represents the in-flight fetch
# wait outside lock to avoid blocking others
try:
return await fut
except asyncio.CancelledError:
# Caller chose to cancel waiting; do not cancel the underlying fetch.
# Re-raise so caller sees cancellation, but fetch continues to populate cache.
raise
# Example usage:
# async def main():
# c = SingleFlightCache(ttl=2.0)
# results = await asyncio.gather(*(c.get("a") for _ in range(10)))
# print(results)
