Lukas
53cdd63b6a
Use NoOpSessionCache in backend/app/main.py and dynamically switch cache implementation in backend/app/dependencies.py so disabled cache mode remains safe while get_session_cache always returns a valid object.
74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
"""Pluggable session cache abstraction.
|
|
|
|
This module defines a cache interface for authenticated sessions and a default
|
|
process-local in-memory implementation. The backend can swap the cache
|
|
implementation without changing the authentication dependency logic.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import TYPE_CHECKING, Protocol
|
|
|
|
if TYPE_CHECKING: # pragma: no cover
|
|
from app.models.auth import Session
|
|
|
|
|
|
class SessionCache(Protocol):
|
|
"""Interface for session token validation cache backends."""
|
|
|
|
def get(self, token: str) -> Session | None:
|
|
"""Return the cached session for *token*, or ``None`` if missing."""
|
|
|
|
def set(self, token: str, session: Session, ttl_seconds: float) -> None:
|
|
"""Cache the validated *session* for *token* for *ttl_seconds*."""
|
|
|
|
def invalidate(self, token: str) -> None:
|
|
"""Remove *token* from the cache if it exists."""
|
|
|
|
def clear(self) -> None:
|
|
"""Remove all entries from the cache."""
|
|
|
|
|
|
class InMemorySessionCache:
|
|
"""A process-local session cache implementation."""
|
|
|
|
def __init__(self) -> None:
|
|
self._entries: dict[str, tuple[Session, float]] = {}
|
|
|
|
def get(self, token: str) -> Session | None:
|
|
entry = self._entries.get(token)
|
|
if entry is None:
|
|
return None
|
|
|
|
session, expires_at = entry
|
|
if time.monotonic() >= expires_at:
|
|
self._entries.pop(token, None)
|
|
return None
|
|
return session
|
|
|
|
def set(self, token: str, session: Session, ttl_seconds: float) -> None:
|
|
self._entries[token] = (session, time.monotonic() + ttl_seconds)
|
|
|
|
def invalidate(self, token: str) -> None:
|
|
self._entries.pop(token, None)
|
|
|
|
def clear(self) -> None:
|
|
self._entries.clear()
|
|
|
|
|
|
class NoOpSessionCache:
|
|
"""A no-op session cache used when caching is disabled."""
|
|
|
|
def get(self, token: str) -> Session | None:
|
|
return None
|
|
|
|
def set(self, token: str, session: Session, ttl_seconds: float) -> None:
|
|
return None
|
|
|
|
def invalidate(self, token: str) -> None:
|
|
return None
|
|
|
|
def clear(self) -> None:
|
|
return None
|