Extract cache/backend.py to mizan-core (Tier A)

cache/backend.py is pure framework-agnostic key-value abstraction —
CacheBackend Protocol, MemoryCache, RedisCache. No Django imports.
Moves to cores/mizan-python/src/mizan_core/cache/backend.py with no
content changes; mizan-django re-imports.

Verified: mizan-core 15/15, mizan-django 348 pass / 21 skip.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-06 14:23:01 -04:00
parent 37e61c646b
commit 9150cdc5ee
3 changed files with 7 additions and 7 deletions

View File

@@ -0,0 +1,115 @@
"""
Cache backends — MemoryCache (testing) and RedisCache (production).
Simple key-value stores. No reverse indexes. Cache keys are derived
from HMAC, so scoped purge just recomputes the key and deletes it.
Broad purge uses key-prefix scan (rare operation).
"""
from __future__ import annotations
from typing import Protocol
class CacheBackend(Protocol):
"""Interface that all Mizan cache backends implement."""
def get(self, key: str) -> bytes | None: ...
def set(self, key: str, value: bytes) -> None: ...
def delete(self, key: str) -> bool: ...
def delete_by_prefix(self, prefix: str) -> int: ...
def clear(self) -> None: ...
class MemoryCache:
"""
In-memory cache backend for testing.
Uses a Python dict. No persistence, no cross-process sharing.
"""
def __init__(self) -> None:
self._store: dict[str, bytes] = {}
def get(self, key: str) -> bytes | None:
return self._store.get(key)
def set(self, key: str, value: bytes) -> None:
self._store[key] = value
def delete(self, key: str) -> bool:
if key in self._store:
del self._store[key]
return True
return False
def delete_by_prefix(self, prefix: str) -> int:
to_delete = [k for k in self._store if k.startswith(prefix)]
for k in to_delete:
del self._store[k]
return len(to_delete)
def clear(self) -> None:
self._store.clear()
class RedisCache:
"""
Redis-backed cache backend for production.
Simple GET/SET/DEL. No reverse indexes. Scoped purge recomputes
the HMAC key and deletes directly. Broad purge uses SCAN.
"""
DEFAULT_TTL = 86400 # 24h safety-net
def __init__(
self,
redis_url: str,
prefix: str = "mizan:",
ttl: int | None = None,
) -> None:
try:
import redis as redis_lib
except ImportError:
raise ImportError(
"Redis is required for Mizan's cache backend. "
"Install it with: pip install mizan[cache]"
)
self._client = redis_lib.from_url(
redis_url,
socket_connect_timeout=5,
socket_timeout=5,
health_check_interval=30,
retry_on_timeout=True,
max_connections=50,
)
self._prefix = prefix
self._ttl = ttl if ttl is not None else self.DEFAULT_TTL
def _key(self, key: str) -> str:
return f"{self._prefix}{key}"
def get(self, key: str) -> bytes | None:
return self._client.get(self._key(key))
def set(self, key: str, value: bytes) -> None:
self._client.set(self._key(key), value, ex=self._ttl)
def delete(self, key: str) -> bool:
return self._client.unlink(self._key(key)) > 0
def delete_by_prefix(self, prefix: str) -> int:
pattern = f"{self._prefix}{prefix}*"
count = 0
cursor = 0
while True:
cursor, keys = self._client.scan(cursor, match=pattern, count=1000)
if keys:
count += self._client.unlink(*keys)
if cursor == 0:
break
return count
def clear(self) -> None:
self.delete_by_prefix("")