perf(session): write on lookup only when sliding expiration renews

CLAUDE.md

@@ -69,6 +69,7 @@ The library has four independent subsystems:
 - Optional subsystem, activated via `SessionMiddleware` and `SessionManagerProxy`.
 - `SessionManager` handles create/get/update/delete/invalidate/regenerate operations. It stores `Session` Pydantic models serialized as JSON, wrapped in `CacheEntry` for backend compatibility.
 - Token signing: `simple` format uses HMAC-SHA256 (`SecurityManager`); `jwt` format uses PyJWT (optional dependency `fastapi-cachex[jwt]`).
+- `get_session()` saves only when sliding expiration renewed the session (or with `touch=True`), so the stored `last_accessed` is the last write, not the last lookup. Session entries use the constant fingerprint `"session"`; nothing compares it.
 - Session token is passed via custom header (`X-Session-Token` by default) or `Authorization: Bearer` token.
 - `SessionManagerProxy` mirrors the `BackendProxy` pattern for managing the `SessionManager` singleton.
 - Key FastAPI dependencies: `get_session`, `require_session`, `get_optional_session` (in `session/dependencies.py`).

docs/SESSION.md

@@ -666,6 +666,13 @@ signature or binding mismatch), `SessionNotFoundError`, `SessionInvalidError`
 Since 0.3.8, `SessionError` derives from `CacheXError`, so `except CacheXError`
 catches session errors too.
 
+`get_session()` writes to the backend only when sliding expiration renewed the session, so a
+request that only reads its session costs a single backend read. The returned
+session's `last_accessed` is the current time, but the stored value is updated
+only when the session is next written (created, modified, renewed or
+regenerated). Pass `touch=True` to save it on every lookup. Before 0.3.8 every
+lookup saved the session.
+
 Every method with its signature is in the generated
 [Session API reference](api/session.md).
 

fastapi_cachex/session/manager.py

@@ -26,6 +26,10 @@
 
 logger = logging.getLogger(__name__)
 
+# Session entries are never compared by fingerprint, so a constant saves
+# hashing the payload on every write.
+_SESSION_FINGERPRINT = "session"
+
 
 def _now() -> datetime:
     return datetime.now(timezone.utc)
@@ -181,13 +185,23 @@ async def get_session(
         token_string: str,
         ip_address: str | None = None,
         user_agent: str | None = None,
+        *,
+        touch: bool = False,
     ) -> tuple[Session, str | None]:
         """Retrieve and validate a session.
 
+        A lookup writes to the backend only when sliding expiration renewed
+        the session, since that is the only change to its expiry or TTL. The
+        returned session's ``last_accessed`` is always the current time, but
+        it is stored only when the session is next written, unless ``touch``
+        is set.
+
         Args:
             token_string: Session token string
             ip_address: Current request IP address
             user_agent: Current request User-Agent
+            touch: Save the session even when it was not renewed, so the
+                stored ``last_accessed`` is exact
 
         Returns:
             Tuple of (session, new_token_string). new_token_string is non-None
@@ -297,7 +311,8 @@ async def get_session(
                     self.config.session_ttl,
                 )
 
-        await self._save_session(session)
+        if renewed_token is not None or touch:
+            await self._save_session(session)
 
         return session, renewed_token
 
@@ -486,7 +501,7 @@ async def _save_session(self, session: Session) -> None:
             ttl = max(ttl, 1)  # Ensure at least 1 second
 
         entry = CacheEntry(
-            fingerprint=self.security.hash_data(payload),
+            fingerprint=_SESSION_FINGERPRINT,
             content=payload.encode("utf-8"),
         )
         await self.backend.set(

i18n/zh-TW/docs/SESSION.md

@@ -540,6 +540,8 @@ session, new_token = await manager.regenerate_session_id(session)
 
 `SessionManager(backend, config, token_serializer=None)` 處理整個生命週期：`create_session()`／`create_anonymous_session()` 回傳 `(session, token)`；`get_session()` 回傳 `(session, renewed_token)`，其中 `renewed_token` 只有在滑動過期更新了權杖時才會有值，並應傳回給用戶端。`get_session()` 失敗時會拋出 `SessionError` 的子類別：`SessionTokenError`（權杖格式錯誤）、`SessionSecurityError`（簽章錯誤或綁定不符）、`SessionNotFoundError`、`SessionInvalidError`（Session 不是啟用狀態）或 `SessionExpiredError`（超過 TTL 或絕對逾時）。從 0.3.8 起，`SessionError` 繼承自 `CacheXError`，因此 `except CacheXError` 也會捕捉 Session 錯誤。
 
+`get_session()` 只有在滑動過期更新了 Session 時才寫入後端，因此只讀取 Session 的請求只需一次後端讀取。回傳的 Session 中 `last_accessed` 是目前時間，但儲存的值只會在 Session 下一次被寫入（建立、修改、更新或重新產生）時更新。傳入 `touch=True` 可在每次查詢時都儲存它。0.3.8 之前，每次查詢都會儲存 Session。
+
 每個方法及其簽名請見自動產生的 [Session API 參考](https://fastapi-cachex.readthedocs.io/en/latest/api/session/)（英文）。
 
 ## 依賴項 {#dependencies}

tests/session/test_lookup_writes.py

@@ -0,0 +1,165 @@
+"""A session lookup writes to the backend only when something needs storing (#115).
+
+Before 0.3.8 every ``get_session()`` re-saved the session to record
+``last_accessed``, so each authenticated request cost a write, and one that
+also modified the session cost two.
+"""
+
+from datetime import datetime
+from datetime import timedelta
+from datetime import timezone
+from typing import Annotated
+
+import pytest
+from fastapi import Depends
+from fastapi import FastAPI
+from fastapi import Request
+from fastapi.testclient import TestClient
+
+from fastapi_cachex.backends.memory import MemoryBackend
+from fastapi_cachex.session.config import SessionConfig
+from fastapi_cachex.session.dependencies import require_session
+from fastapi_cachex.session.manager import SessionManager
+from fastapi_cachex.session.middleware import FastAPICacheXSessionMiddleware
+from fastapi_cachex.session.middleware import SessionMiddleware
+from fastapi_cachex.session.models import Session
+from fastapi_cachex.session.models import SessionUser
+from fastapi_cachex.types import CacheEntry
+
+
+class SpyBackend(MemoryBackend):
+    """A memory backend that records every key passed to ``set``."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.writes: list[str] = []
+
+    async def set(self, key: str, value: CacheEntry, ttl: int | None = None) -> None:
+        self.writes.append(key)
+        await super().set(key, value, ttl=ttl)
+
+
+def _manager(**overrides: object) -> tuple[SessionManager, SpyBackend]:
+    backend = SpyBackend()
+    config = SessionConfig(secret_key="a" * 32, **overrides)
+    return SessionManager(backend, config), backend
+
+
+async def _stored(manager: SessionManager, session_id: str) -> Session:
+    stored = await manager._load_session(session_id)
+    assert stored is not None
+    return stored
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sliding_expiration", [True, False])
+async def test_plain_lookup_does_not_write(sliding_expiration: bool) -> None:
+    """A lookup that renews nothing leaves the backend untouched."""
+    manager, backend = _manager(sliding_expiration=sliding_expiration)
+    session, token = await manager.create_session(SessionUser(user_id="u1"))
+    backend.writes.clear()
+
+    loaded, renewed = await manager.get_session(token)
+
+    assert renewed is None
+    assert backend.writes == []
+    assert loaded.session_id == session.session_id
+
+
+@pytest.mark.asyncio
+async def test_sliding_renewal_writes_once_and_extends_the_stored_expiry() -> None:
+    """A renewal must reach the backend, or its TTL would not move."""
+    manager, backend = _manager(session_ttl=3600, sliding_threshold=0.5)
+    session, token = await manager.create_session(SessionUser(user_id="u1"))
+    session.expires_at = datetime.now(timezone.utc) + timedelta(seconds=100)
+    await manager.update_session(session)
+    backend.writes.clear()
+
+    loaded, renewed = await manager.get_session(token)
+
+    assert renewed is not None
+    assert len(backend.writes) == 1
+    stored = await _stored(manager, session.session_id)
+    assert stored.expires_at == loaded.expires_at
+    assert stored.expires_at > datetime.now(timezone.utc) + timedelta(seconds=3000)  # type: ignore[operator]
+
+
+@pytest.mark.asyncio
+async def test_touch_stores_last_accessed() -> None:
+    """``touch=True`` keeps the stored ``last_accessed`` exact."""
+    manager, backend = _manager()
+    session, token = await manager.create_session(SessionUser(user_id="u1"))
+    backend.writes.clear()
+
+    untouched, _ = await manager.get_session(token)
+    assert backend.writes == []
+    assert (await _stored(manager, session.session_id)).last_accessed < (
+        untouched.last_accessed
+    )
+
+    touched, _ = await manager.get_session(token, touch=True)
+    assert len(backend.writes) == 1
+    stored = await _stored(manager, session.session_id)
+    assert stored.last_accessed == touched.last_accessed
+
+
+@pytest.mark.asyncio
+async def test_session_entries_use_a_constant_fingerprint() -> None:
+    """Nothing compares session fingerprints, so the payload is not hashed."""
+    manager, backend = _manager()
+    session, _ = await manager.create_session(SessionUser(user_id="u1"))
+
+    entry = await backend.get(manager._get_backend_key(session.session_id))
+
+    assert entry is not None
+    assert entry.fingerprint == "session"
+
+
+def test_middleware_reads_without_writing_and_writes_changes_once() -> None:
+    """Reading ``request.session`` costs no write; changing it costs exactly one."""
+    manager, backend = _manager()
+    app = FastAPI()
+    app.add_middleware(FastAPICacheXSessionMiddleware, session_manager=manager)
+
+    @app.get("/read")
+    async def read(request: Request) -> dict[str, object]:
+        return {"count": request.session.get("count", 0)}
+
+    @app.get("/bump")
+    async def bump(request: Request) -> dict[str, object]:
+        request.session["count"] = request.session.get("count", 0) + 1
+        return {"count": request.session["count"]}
+
+    client = TestClient(app)
+    assert client.get("/bump").json() == {"count": 1}  # creates the session
+    backend.writes.clear()
+
+    assert client.get("/read").json() == {"count": 1}
+    assert backend.writes == []
+
+    assert client.get("/bump").json() == {"count": 2}
+    assert len(backend.writes) == 1
+    assert client.get("/read").json() == {"count": 2}
+
+
+@pytest.mark.asyncio
+@pytest.mark.filterwarnings("ignore::DeprecationWarning")
+async def test_header_middleware_lookup_does_not_write() -> None:
+    """The deprecated header middleware loads sessions the same way."""
+    manager, backend = _manager()
+    app = FastAPI()
+    app.add_middleware(SessionMiddleware, session_manager=manager)
+
+    @app.get("/me")
+    async def me(
+        session: Annotated[Session, Depends(require_session)],
+    ) -> dict[str, object]:
+        return {"user": session.user.user_id if session.user else None}
+
+    _, token = await manager.create_session(SessionUser(user_id="u1"))
+    backend.writes.clear()
+
+    response = TestClient(app).get("/me", headers={"X-Session-Token": token})
+
+    assert response.json() == {"user": "u1"}
+    assert backend.writes == []