Add storage and cache implementations

FileTokenStorage (JSON file, chmod 0o600), MemoryTokenStorage (dict-based),
FileCache (per-key files with TTL), MemoryCache (dict with TTL) --
Python equivalents of the PHP FileTokenStorage, SessionTokenStorage,
FileCache, and ArrayCache.

Co-Authored-By: claude-flow <ruv@ruv.net>

Author: Allow2CEO

allow2_service/cache/__init__.py

@@ -0,0 +1,9 @@
+"""Cache implementations."""
+
+from allow2_service.cache.file_cache import FileCache
+from allow2_service.cache.memory_cache import MemoryCache
+
+__all__ = [
+    "FileCache",
+    "MemoryCache",
+]

allow2_service/cache/file_cache.py

@@ -0,0 +1,81 @@
+"""File-based permission cache.
+
+Each cache entry is stored as a separate JSON file in the cache directory.
+Suitable for single-server deployments.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import time
+from typing import Optional
+
+from allow2_service.cache_interface import CacheInterface
+
+
+class FileCache(CacheInterface):
+    """File-based permission cache.
+
+    Each cache entry is stored as a separate JSON file in the cache
+    directory. Expired entries are removed lazily on read.
+
+    Args:
+        cache_dir: Directory for cache files. Created automatically if missing.
+    """
+
+    def __init__(self, cache_dir: str) -> None:
+        self._cache_dir = cache_dir
+        if not os.path.isdir(self._cache_dir):
+            os.makedirs(self._cache_dir, mode=0o700, exist_ok=True)
+
+    def get(self, key: str) -> Optional[str]:
+        """Retrieve a cached value by key."""
+        path = self._key_to_path(key)
+
+        if not os.path.exists(path):
+            return None
+
+        try:
+            with open(path, "r", encoding="utf-8") as f:
+                entry = json.load(f)
+        except (json.JSONDecodeError, OSError):
+            self._safe_unlink(path)
+            return None
+
+        if "expires_at" not in entry or int(time.time()) >= entry["expires_at"]:
+            self._safe_unlink(path)
+            return None
+
+        return entry.get("value")
+
+    def set(self, key: str, value: str, ttl: int = 60) -> None:
+        """Store a value in cache."""
+        path = self._key_to_path(key)
+
+        entry = json.dumps({
+            "value": value,
+            "expires_at": int(time.time()) + ttl,
+        })
+
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(entry)
+
+    def delete(self, key: str) -> None:
+        """Delete a cached value."""
+        path = self._key_to_path(key)
+        self._safe_unlink(path)
+
+    def _key_to_path(self, key: str) -> str:
+        """Convert a cache key to a safe filesystem path."""
+        safe_key = re.sub(r"[^a-zA-Z0-9_\-]", "_", key)
+        return os.path.join(self._cache_dir, safe_key + ".json")
+
+    @staticmethod
+    def _safe_unlink(path: str) -> None:
+        """Remove a file if it exists, suppressing errors."""
+        try:
+            os.unlink(path)
+        except OSError:
+            pass

allow2_service/cache/memory_cache.py

@@ -0,0 +1,46 @@
+"""In-memory cache that lives for the duration of the process.
+
+No persistence between restarts. Useful for testing or when you
+only need to deduplicate checks within a single request lifecycle.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Dict, Optional
+
+from allow2_service.cache_interface import CacheInterface
+
+
+class MemoryCache(CacheInterface):
+    """In-memory dict-based cache with TTL support.
+
+    Each entry stores a value and an ``expires_at`` timestamp.
+    Expired entries are removed lazily on read.
+    """
+
+    def __init__(self) -> None:
+        self._store: Dict[str, Dict[str, object]] = {}
+
+    def get(self, key: str) -> Optional[str]:
+        """Retrieve a cached value by key."""
+        if key not in self._store:
+            return None
+
+        entry = self._store[key]
+        if int(time.time()) >= int(entry["expires_at"]):  # type: ignore[arg-type]
+            del self._store[key]
+            return None
+
+        return str(entry["value"])
+
+    def set(self, key: str, value: str, ttl: int = 60) -> None:
+        """Store a value in cache."""
+        self._store[key] = {
+            "value": value,
+            "expires_at": int(time.time()) + ttl,
+        }
+
+    def delete(self, key: str) -> None:
+        """Delete a cached value."""
+        self._store.pop(key, None)

allow2_service/storage/__init__.py

@@ -0,0 +1,9 @@
+"""Token storage implementations."""
+
+from allow2_service.storage.file_storage import FileTokenStorage
+from allow2_service.storage.memory_storage import MemoryTokenStorage
+
+__all__ = [
+    "FileTokenStorage",
+    "MemoryTokenStorage",
+]

allow2_service/storage/file_storage.py

@@ -0,0 +1,89 @@
+"""File-based JSON token storage.
+
+Stores all tokens in a single JSON file. Suitable for development
+and single-server deployments. Not recommended for high-concurrency
+production use.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from typing import Any, Dict, Optional
+
+from allow2_service.models.oauth_tokens import OAuthTokens
+from allow2_service.token_storage import TokenStorageInterface
+
+
+class FileTokenStorage(TokenStorageInterface):
+    """File-based JSON token storage.
+
+    Stores all tokens in a single JSON file with ``chmod 0600``
+    for security.
+
+    Args:
+        file_path: Path to the JSON storage file.
+    """
+
+    def __init__(self, file_path: str) -> None:
+        self._file_path = file_path
+        self._data: Optional[Dict[str, Any]] = None
+
+    def store(self, user_id: str, tokens: OAuthTokens) -> None:
+        """Store tokens for a user."""
+        data = self._load_all()
+        data[user_id] = tokens.to_dict()
+        self._save_all(data)
+
+    def retrieve(self, user_id: str) -> Optional[OAuthTokens]:
+        """Retrieve tokens for a user, or None if none stored."""
+        data = self._load_all()
+
+        if user_id not in data:
+            return None
+
+        return OAuthTokens.from_dict(data[user_id])
+
+    def delete(self, user_id: str) -> None:
+        """Delete tokens for a user."""
+        data = self._load_all()
+        data.pop(user_id, None)
+        self._save_all(data)
+
+    def exists(self, user_id: str) -> bool:
+        """Check whether tokens exist for a user."""
+        data = self._load_all()
+        return user_id in data
+
+    def _load_all(self) -> Dict[str, Any]:
+        """Load all tokens from disk."""
+        if self._data is not None:
+            return self._data
+
+        if not os.path.exists(self._file_path):
+            self._data = {}
+            return self._data
+
+        try:
+            with open(self._file_path, "r", encoding="utf-8") as f:
+                self._data = json.load(f)
+        except (json.JSONDecodeError, OSError):
+            self._data = {}
+
+        return self._data
+
+    def _save_all(self, data: Dict[str, Any]) -> None:
+        """Persist all tokens to disk."""
+        self._data = data
+
+        dir_path = os.path.dirname(self._file_path)
+        if dir_path and not os.path.isdir(dir_path):
+            os.makedirs(dir_path, mode=0o700, exist_ok=True)
+
+        with open(self._file_path, "w", encoding="utf-8") as f:
+            json.dump(data, f, indent=2)
+
+        try:
+            os.chmod(self._file_path, 0o600)
+        except OSError:
+            pass  # chmod may not be available on all platforms

allow2_service/storage/memory_storage.py

@@ -0,0 +1,42 @@
+"""In-memory token storage.
+
+Suitable for testing or single-process development. Tokens are lost
+when the process exits.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, Optional
+
+from allow2_service.models.oauth_tokens import OAuthTokens
+from allow2_service.token_storage import TokenStorageInterface
+
+
+class MemoryTokenStorage(TokenStorageInterface):
+    """In-memory dict-based token storage.
+
+    No persistence between process restarts. Useful for testing or
+    simple single-process applications.
+    """
+
+    def __init__(self) -> None:
+        self._store: Dict[str, Dict[str, object]] = {}
+
+    def store(self, user_id: str, tokens: OAuthTokens) -> None:
+        """Store tokens for a user."""
+        self._store[user_id] = tokens.to_dict()
+
+    def retrieve(self, user_id: str) -> Optional[OAuthTokens]:
+        """Retrieve tokens for a user, or None if none stored."""
+        if user_id not in self._store:
+            return None
+
+        return OAuthTokens.from_dict(self._store[user_id])  # type: ignore[arg-type]
+
+    def delete(self, user_id: str) -> None:
+        """Delete tokens for a user."""
+        self._store.pop(user_id, None)
+
+    def exists(self, user_id: str) -> bool:
+        """Check whether tokens exist for a user."""
+        return user_id in self._store