Skip to content

Sessions

Session management: a store protocol, an in-memory implementation, cookie signing helpers, and the session_cookie guard that ties them together.

Session

@dataclass(slots=True)
class Session:
    id: str
    user_id: str
    data: dict[str, Any] = field(default_factory=dict)
    created_at: float = field(default_factory=time.time)
    expires_at: float | None = None

SessionStore (Protocol)

A session store is a three-method async protocol:

class SessionStore(Protocol):
    async def create(
        self,
        *,
        user_id: str,
        data: dict[str, Any] | None = None,
        ttl_seconds: int | None = None,
    ) -> Session: ...

    async def get(self, session_id: str) -> Session | None: ...

    async def delete(self, session_id: str) -> None: ...

Implement it against Redis, Postgres, or anything else shared across workers.

InMemorySessionStore

from lauren_guards import InMemorySessionStore

store = InMemorySessionStore()

session = await store.create(user_id="u1", data={"role": "admin"}, ttl_seconds=3600)
loaded = await store.get(session.id)
assert loaded.user_id == "u1"
await store.delete(session.id)

Process-local and thread-safe. Fine for development, tests, and single-worker deployments. Sessions expire lazily: get on an expired session removes it and returns None.

sign_cookie / verify_cookie implement HMAC-SHA256 signing for cookie values:

from lauren_guards import sign_cookie, verify_cookie

secret = "my-signing-secret"
signed = sign_cookie("user-123", secret=secret)
# e.g. "user-123.<hex-hmac>"

value = verify_cookie(signed, secret=secret)
assert value == "user-123"   # str | None

assert verify_cookie(signed, secret="wrong-secret") is None
assert verify_cookie("tampered.value", secret=secret) is None
  • sign_cookie(session_id, *, secret) -> str — returns "<id>.<sig>"; raises ValueError if the id contains a ..
  • verify_cookie(value, *, secret) -> str | None — returns the original value on success, None on tamper / wrong secret / malformed input. It never raises.
from lauren_guards import InMemorySessionStore, session_cookie, sign_cookie

store = InMemorySessionStore()
SessionGuard = session_cookie(store=store, secret="my-secret")

In a login handler, create the session and set the signed cookie; the guard reads the cookie and re-hydrates the Session from the store on subsequent requests:

async def login(request) -> Response:
    session = await store.create(user_id=request.username, ttl_seconds=3600)
    signed = sign_cookie(session.id, secret="my-secret")
    return Response.json({"ok": True}).with_cookie(
        "lauren_session", signed, http_only=True, secure=True, same_site="lax",
    )

See the Session Cookie guard for the full guard documentation.

Production backends

Swap InMemorySessionStore for a shared store by implementing the SessionStore protocol — e.g. a Redis-backed store:

import json
import redis.asyncio as redis
from lauren_guards import SessionStore, Session

class RedisSessionStore(SessionStore):
    def __init__(self, redis_url: str, ttl: int = 3600) -> None:
        self._r = redis.from_url(redis_url)
        self._ttl = ttl

    async def create(self, *, user_id, data=None, ttl_seconds=None):
        session = Session(
            id=secrets.token_urlsafe(24),
            user_id=user_id,
            data=data or {},
            expires_at=time.time() + (ttl_seconds or self._ttl),
        )
        await self._r.setex(
            f"sessions:{session.id}", ttl_seconds or self._ttl,
            json.dumps({"user_id": session.user_id, "data": session.data}),
        )
        return session

    async def get(self, session_id: str) -> Session | None:
        raw = await self._r.get(f"sessions:{session_id}")
        if raw is None:
            return None
        payload = json.loads(raw)
        session = Session(id=session_id, **payload)
        if session.expires_at and session.expires_at < time.time():
            await self.delete(session_id)
            return None
        return session

    async def delete(self, session_id: str) -> None:
        await self._r.delete(f"sessions:{session_id}")