Replace process-local session cache with pluggable session cache backend
This commit is contained in:
57
backend/app/utils/session_cache.py
Normal file
57
backend/app/utils/session_cache.py
Normal file
@@ -0,0 +1,57 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user