Sessions

The session mapping and the server-side stores that back it.

Session

Bases: dict[str, Any]

The request session - a dict that knows when it has changed.

permanent property writable

permanent: bool

Whether the session cookie should use the longer lifetime.

Backed by the reserved _permanent key, so the flag persists in the cookie across requests and toggling it counts as a session mutation.

regenerate_id

regenerate_id() -> None

Request a fresh server-side session id on the next response.

Call this at a privilege boundary - login, role change - so a pre-existing (possibly attacker-planted) session id cannot be replayed against the now-elevated session: the session-fixation defence. It marks the session modified so the rotation is written back. Harmless with cookie-only sessions, which carry no server-side id to rotate.

clear

clear() -> None

Remove every key and mark the session modified.

pop

pop(key: Any, *default: Any) -> Any

Remove key and return its value, marking the session modified on removal.

popitem

popitem() -> Any

Remove and return the last (key, value) pair, marking the session modified.

setdefault

setdefault(key: Any, default: Any = None) -> Any

Insert default under key when absent, marking the session modified on insert.

update

update(*args: Any, **kwargs: Any) -> None

Merge keys into the session, marking it modified unless the input is empty.

SessionStore

Server-side session backend interface.

A concrete store persists session payloads keyed by an opaque session id; ServerSessionMiddleware drives it. The methods are async so a network-backed store (Redis, a database) can implement them without blocking the event loop - the bundled InMemorySessionStore satisfies the contract without any real awaiting.

read async

read(session_id: str) -> dict[str, Any] | None

Return the stored payload for session_id, or None.

None when the session is absent, expired, or has been revoked.

write async

write(session_id: str, data: dict[str, Any], max_age: int) -> None

Persist data under session_id, to expire after max_age seconds.

delete async

delete(session_id: str) -> None

Revoke session_id - a later read of it must return None.

replace async

replace(session_id: str, data: dict[str, Any], max_age: int) -> bool

Write data for session_id only if it still exists.

Returns True on success, False when the id is absent - it was revoked or expired. This is the race-safe write the middleware uses for an already-stored session, so a request still in flight cannot resurrect a session a concurrent delete removed.

The default is a non-atomic read-then-write; a store with an atomic conditional write (Redis SET ... XX, a DB UPDATE) should override this to close the check-then-write window.

touch async

touch(session_id: str, max_age: int) -> bool

Extend the expiry of an existing entry without rewriting its payload.

Returns True when the id existed and its TTL was refreshed, False when it was absent (revoked or expired). This is the sliding-expiry write ServerSessionMiddleware uses on a read-only access, so an idle session stays alive without round-tripping its full payload.

The default reads then rewrites the payload; a store with a native TTL-refresh primitive (Redis EXPIRE, a DB UPDATE ... expires_at) should override this to avoid moving the payload.

InMemorySessionStore

Bases: SessionStore

A process-local SessionStore - a dict with per-entry expiry.

Fine for a single-process app and for tests. It does not share state across workers, so a multi-worker deployment needs a shared backend (e.g. Redis) implementing the SessionStore interface.

Sized, iterable and containment-testable, so it is falsy when it holds no live session - as any empty collection is. Test store is not None to ask whether a store is configured; if store: asks whether it currently holds anything, which at process start it does not.

read async

read(session_id: str) -> dict[str, Any] | None

Return a copy of the stored payload, or None when absent or expired.

write async

write(session_id: str, data: dict[str, Any], max_age: int) -> None

Store a copy of data under session_id, expiring after max_age seconds.

delete async

delete(session_id: str) -> None

Drop session_id from the store. No-op if not present.

replace async

replace(session_id: str, data: dict[str, Any], max_age: int) -> bool

Write data only when session_id still exists and is unexpired.

touch async

touch(session_id: str, max_age: int) -> bool

Refresh the expiry of an existing, unexpired entry without copying its payload.

sweep_expired

sweep_expired() -> int

Drop every expired entry and return how many were removed.

Callers that want deterministic eviction (e.g. a background task on a known cadence) can call this directly rather than relying on the probabilistic sweep that fires from write / replace.

clear

clear() -> int

Drop every session and return how many were removed.

The sync counterpart to delete for the whole store - what "log everyone out" needs after a key rotation or a breach. Expired entries that no sweep has reached are counted as removed too, since they were occupying the store.

expires_at

expires_at(session_id: str) -> float | None

Return when session_id expires as a Unix timestamp, or None.

None means the id is absent or already past its expiry, matching what read would say - an entry the lazy sweep has not reached yet is gone as far as every accessor is concerned.

This is what sliding expiry is observable through: the payload does not change when a TTL is refreshed, so read cannot show that touch did anything.