From c888b50867a817f0ec6de74d5f1f62cb84e8b2dd Mon Sep 17 00:00:00 2001 From: Silvano Cerza Date: Fri, 14 Aug 2026 15:04:26 +0200 Subject: [PATCH] Add cache TTL --- .env.example | 3 ++ README.md | 1 + src/ipinfo_mcp/cache.py | 36 ++++++++++++++--- src/ipinfo_mcp/server.py | 9 +++-- tests/test_cache.py | 83 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 123 insertions(+), 9 deletions(-) diff --git a/.env.example b/.env.example index 1ed4467..4c2fbe3 100644 --- a/.env.example +++ b/.env.example @@ -7,3 +7,6 @@ IPINFO_TOKEN= # Base URL for the legacy ipinfo.io endpoints, e.g. /me (default: https://ipinfo.io) # IPINFO_LEGACY_BASE_URL=https://ipinfo.io + +# How long cached IP results stay fresh, in seconds (default: 3600) +# IPINFO_CACHE_TTL=3600 diff --git a/README.md b/README.md index f39e216..2f84a9e 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,7 @@ IPINFO_TRANSPORT=http IPINFO_HOST=0.0.0.0 IPINFO_PORT=8000 uv run ipinfo-mcp-ser | `IPINFO_TOKEN` | | API token | | `IPINFO_API_BASE_URL` | `https://api.ipinfo.io` | Base URL for `api.ipinfo.io` endpoints | | `IPINFO_LEGACY_BASE_URL` | `https://ipinfo.io` | Base URL for legacy `ipinfo.io` endpoints (e.g. `/me`) | +| `IPINFO_CACHE_TTL` | `3600` | Seconds a cached IP result stays fresh | | `IPINFO_TRANSPORT` | `stdio` | Transport type (`stdio` or `http`) | | `IPINFO_HOST` | `0.0.0.0` | HTTP host (only for `http` transport) | | `IPINFO_PORT` | `8000` | HTTP port (only for `http` transport) | diff --git a/src/ipinfo_mcp/cache.py b/src/ipinfo_mcp/cache.py index 03d38ed..bb9fb3d 100644 --- a/src/ipinfo_mcp/cache.py +++ b/src/ipinfo_mcp/cache.py @@ -1,10 +1,14 @@ import hashlib +import time from ipinfo_mcp.types import LiteResponse, LookupResponse, ResproxyResponse # Any single-IP response that can be cached CachedResponse = LiteResponse | LookupResponse | ResproxyResponse | dict[str, object] +# How long an entry stays fresh, in seconds. +DEFAULT_TTL = 3600.0 + class IPCache: """ @@ -15,10 +19,16 @@ class IPCache: Entries are scoped also using the token that fetched them. The cache is process wide, so if we don't use the token as part of the key we would hand one token's results to another. + + Entries expire after ttl seconds. Expiry is checked on read, and an expired + entry is dropped as we find it, so nothing sweeps the store in the background. """ - def __init__(self) -> None: - self._store: dict[tuple[str, str, str], CachedResponse] = {} + def __init__(self, ttl: float = DEFAULT_TTL) -> None: + self._ttl = ttl + # Each entry is (stored_at, data), stored_at coming from a monotonic clock + # so that a system clock change can't make an entry look fresh forever. + self._store: dict[tuple[str, str, str], tuple[float, CachedResponse]] = {} @staticmethod def _hash_token(token: str) -> str: @@ -29,13 +39,26 @@ def _hash_token(token: str) -> str: """ return hashlib.sha256(token.encode()).hexdigest()[:32] + def _get_fresh(self, key: tuple[str, str, str]) -> CachedResponse | None: + """Read an entry by key, dropping and reporting it as a miss if it expired.""" + entry = self._store.get(key) + if entry is None: + return None + + stored_at, data = entry + if time.monotonic() > stored_at + self._ttl: + del self._store[key] + return None + + return data + def get(self, token: str, namespace: str, ip: str) -> CachedResponse | None: """Get this token's cached data for an IP in a namespace. Returns None on miss.""" - return self._store.get((self._hash_token(token), namespace, ip)) + return self._get_fresh((self._hash_token(token), namespace, ip)) def put(self, token: str, namespace: str, ip: str, data: CachedResponse) -> None: """Store data for an IP in a namespace, scoped to this token.""" - self._store[(self._hash_token(token), namespace, ip)] = data + self._store[(self._hash_token(token), namespace, ip)] = (time.monotonic(), data) def get_many(self, token: str, namespace: str, ips: list[str]) -> tuple[dict[str, CachedResponse], list[str]]: """Look up multiple IPs for this token. Returns (cached_results, cache_misses).""" @@ -43,7 +66,7 @@ def get_many(self, token: str, namespace: str, ips: list[str]) -> tuple[dict[str cached: dict[str, CachedResponse] = {} misses: list[str] = [] for ip in ips: - data = self._store.get((token_hash, namespace, ip)) + data = self._get_fresh((token_hash, namespace, ip)) if data is not None: cached[ip] = data else: @@ -53,5 +76,6 @@ def get_many(self, token: str, namespace: str, ips: list[str]) -> tuple[dict[str def put_many(self, token: str, namespace: str, items: dict[str, CachedResponse]) -> None: """Store multiple IP results in a namespace, scoped to this token.""" token_hash = self._hash_token(token) + stored_at = time.monotonic() for ip, data in items.items(): - self._store[(token_hash, namespace, ip)] = data + self._store[(token_hash, namespace, ip)] = (stored_at, data) diff --git a/src/ipinfo_mcp/server.py b/src/ipinfo_mcp/server.py index 3128b73..c2ea862 100644 --- a/src/ipinfo_mcp/server.py +++ b/src/ipinfo_mcp/server.py @@ -11,7 +11,7 @@ from starlette.responses import HTMLResponse, JSONResponse from starlette.types import ASGIApp, Receive, Scope, Send -from ipinfo_mcp.cache import IPCache +from ipinfo_mcp.cache import DEFAULT_TTL, IPCache from ipinfo_mcp.client import IPinfoClient from ipinfo_mcp.logging import setup_logging from ipinfo_mcp.tools.asn import register_asn @@ -28,6 +28,7 @@ class Settings(TypedDict): api_token: str | None api_base_url: str legacy_base_url: str + cache_ttl: float def _settings() -> Settings: @@ -35,6 +36,7 @@ def _settings() -> Settings: "api_token": os.environ.get("IPINFO_TOKEN"), "api_base_url": os.environ.get("IPINFO_API_BASE_URL", "https://api.ipinfo.io"), "legacy_base_url": os.environ.get("IPINFO_LEGACY_BASE_URL", "https://ipinfo.io"), + "cache_ttl": float(os.environ.get("IPINFO_CACHE_TTL", DEFAULT_TTL)), } @@ -52,10 +54,11 @@ async def lifespan(_: FastMCP) -> AsyncIterator[ContextData]: base_url=settings["api_base_url"], legacy_base_url=settings["legacy_base_url"], ) as client: - cache = IPCache() + cache = IPCache(ttl=settings["cache_ttl"]) logger.info( - "IPinfo MCP server started (token=%s)", + "IPinfo MCP server started (token=%s cache_ttl=%s)", "configured" if settings["api_token"] else "anonymous", + settings["cache_ttl"], ) yield {"client": client, "cache": cache, "api_token": settings["api_token"]} logger.info("IPinfo MCP server stopped") diff --git a/tests/test_cache.py b/tests/test_cache.py index b5501c0..4f06384 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -1,9 +1,32 @@ +import pytest + +from ipinfo_mcp import cache as cache_module from ipinfo_mcp.cache import IPCache TOKEN = "token_a" OTHER_TOKEN = "token_b" +class FakeClock: + """Controllable stand-in for time.monotonic, so TTL tests need no sleeps.""" + + def __init__(self, now: float = 1000.0) -> None: + self.now = now + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +@pytest.fixture +def clock(monkeypatch: pytest.MonkeyPatch) -> FakeClock: + fake = FakeClock() + monkeypatch.setattr(cache_module.time, "monotonic", fake) + return fake + + class TestIPCachePutGet: def test_get_returns_none_for_missing_key(self) -> None: cache = IPCache() @@ -124,3 +147,63 @@ def test_put_many_is_scoped_to_token(self) -> None: cache.put_many(TOKEN, "lite", {"8.8.8.8": {"ip": "8.8.8.8"}}) assert cache.get(OTHER_TOKEN, "lite", "8.8.8.8") is None + + +class TestIPCacheTTL: + def test_entry_is_fresh_before_ttl(self, clock: FakeClock) -> None: + cache = IPCache(ttl=60.0) + cache.put(TOKEN, "lite", "8.8.8.8", {"ip": "8.8.8.8"}) + + clock.advance(59.0) + assert cache.get(TOKEN, "lite", "8.8.8.8") == {"ip": "8.8.8.8"} + + def test_entry_is_fresh_exactly_at_ttl(self, clock: FakeClock) -> None: + cache = IPCache(ttl=60.0) + cache.put(TOKEN, "lite", "8.8.8.8", {"ip": "8.8.8.8"}) + + clock.advance(60.0) + assert cache.get(TOKEN, "lite", "8.8.8.8") == {"ip": "8.8.8.8"} + + def test_entry_expires_after_ttl(self, clock: FakeClock) -> None: + cache = IPCache(ttl=60.0) + cache.put(TOKEN, "lite", "8.8.8.8", {"ip": "8.8.8.8"}) + + clock.advance(60.1) + assert cache.get(TOKEN, "lite", "8.8.8.8") is None + + def test_expired_entry_is_dropped_from_the_store(self, clock: FakeClock) -> None: + """Expiry cleans as it reads, so an expired entry doesn't linger.""" + cache = IPCache(ttl=60.0) + cache.put(TOKEN, "lite", "8.8.8.8", {"ip": "8.8.8.8"}) + + clock.advance(61.0) + _ = cache.get(TOKEN, "lite", "8.8.8.8") + + assert cache._store == {} + + def test_expired_entry_is_a_miss_in_get_many(self, clock: FakeClock) -> None: + cache = IPCache(ttl=60.0) + cache.put(TOKEN, "lite", "8.8.8.8", {"ip": "8.8.8.8"}) + clock.advance(61.0) + cache.put(TOKEN, "lite", "1.1.1.1", {"ip": "1.1.1.1"}) + + cached, misses = cache.get_many(TOKEN, "lite", ["8.8.8.8", "1.1.1.1"]) + assert cached == {"1.1.1.1": {"ip": "1.1.1.1"}} + assert misses == ["8.8.8.8"] + + def test_put_resets_the_ttl(self, clock: FakeClock) -> None: + cache = IPCache(ttl=60.0) + cache.put(TOKEN, "lite", "8.8.8.8", {"old": True}) + + clock.advance(59.0) + cache.put(TOKEN, "lite", "8.8.8.8", {"new": True}) + + clock.advance(59.0) + assert cache.get(TOKEN, "lite", "8.8.8.8") == {"new": True} + + def test_put_many_entries_expire(self, clock: FakeClock) -> None: + cache = IPCache(ttl=60.0) + cache.put_many(TOKEN, "lite", {"8.8.8.8": {"ip": "8.8.8.8"}}) + + clock.advance(61.0) + assert cache.get(TOKEN, "lite", "8.8.8.8") is None