import json
import secrets
import time
from typing import Optional
import redis.asyncio as redis
class SlidingSessionStore:
def __init__(self, client: redis.Redis, idle_ttl: int = 1800, max_lifetime: int = 86400):
self._redis = client
self._idle_ttl = idle_ttl
self._max_lifetime = max_lifetime
def _key(self, token: str) -> str:
return f"session:{token}"
async def create(self, user_id: int) -> str:
token = secrets.token_urlsafe(32)
now = int(time.time())
payload = {
"user_id": user_id,
"created_at": now,
"absolute_deadline": now + self._max_lifetime,
}
await self._redis.setex(self._key(token), self._idle_ttl, json.dumps(payload))
return token
async def touch(self, token: str) -> Optional[dict]:
if not token:
return None
raw = await self._redis.get(self._key(token))
if raw is None:
return None
session = json.loads(raw)
now = int(time.time())
if now >= session["absolute_deadline"]:
await self._redis.delete(self._key(token))
return None
# slide the idle window forward without exceeding the hard deadline
remaining = session["absolute_deadline"] - now
await self._redis.expire(self._key(token), min(self._idle_ttl, remaining))
return session
async def destroy(self, token: str) -> None:
if token:
await self._redis.delete(self._key(token))
from http.cookies import SimpleCookie
from starlette.types import ASGIApp, Receive, Scope, Send
from .session_store import SlidingSessionStore
class SlidingSessionMiddleware:
def __init__(self, app: ASGIApp, store: SlidingSessionStore, cookie_name: str = "session"):
self._app = app
self._store = store
self._cookie_name = cookie_name
def _read_token(self, scope: Scope) -> str:
for name, value in scope.get("headers", []):
if name == b"cookie":
jar = SimpleCookie()
jar.load(value.decode("latin-1"))
morsel = jar.get(self._cookie_name)
if morsel is not None:
return morsel.value
return ""
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http":
await self._app(scope, receive, send)
return
token = self._read_token(scope)
session = await self._store.touch(token)
scope["session"] = session or {}
scope["session_token"] = token
await self._app(scope, receive, send)
from fastapi import APIRouter, HTTPException, Request, Response
from .dependencies import get_store
router = APIRouter()
@router.post("/login")
async def login(request: Request, response: Response):
body = await request.json()
user_id = authenticate(body["email"], body["password"])
if user_id is None:
raise HTTPException(status_code=401, detail="invalid credentials")
store = get_store(request)
token = await store.create(user_id)
response.set_cookie(
"session",
token,
httponly=True,
secure=True,
samesite="lax",
max_age=86400,
)
return {"ok": True}
@router.get("/whoami")
async def whoami(request: Request):
session = request.scope.get("session") or {}
if not session:
raise HTTPException(status_code=401, detail="session expired or missing")
return {"user_id": session["user_id"], "expires_at": session["absolute_deadline"]}
@router.post("/logout")
async def logout(request: Request, response: Response):
store = get_store(request)
await store.destroy(request.scope.get("session_token", ""))
response.delete_cookie("session")
return {"ok": True}
This snippet implements a sliding-window session model where a token stays valid as long as it is used, but expires after a fixed period of inactivity. It is the pattern behind "you were logged out because you were idle for 30 minutes", and it avoids both the annoyance of hard absolute expiry and the risk of tokens that live forever.
In session_store.py, SlidingSessionStore wraps Redis and treats each session as a single key whose TTL is the sliding window. On create, a random token is generated with secrets.token_urlsafe and stored with setex, so Redis itself enforces expiry — no cron job or sweeper is needed. The key trick is in touch: it reads the session, and if present, calls expire to reset the TTL back to the full idle_ttl. That single expire call is what makes the window "slide" forward on every authenticated request. An absolute_deadline is baked into the payload at creation so a session can never be renewed past a hard ceiling, which caps the damage from a stolen-but-active token.
The use of GET followed by a separate EXPIRE introduces a small race, but for session refresh it is acceptable; a stricter version would use a Lua script to make the read-and-extend atomic. Notably touch refuses to extend once absolute_deadline has passed, so idle refresh and lifetime cap are enforced in the same place.
In middleware.py, SlidingSessionMiddleware is a raw ASGI middleware that runs on every request. It extracts the token from the session cookie, calls store.touch, and attaches the resolved session to scope['session'] so downstream handlers can read it without touching Redis again. When a session is missing or expired, the scope value is simply left empty rather than rejecting the request, which keeps public routes working and defers auth decisions to the route layer.
In routes.py, the login route creates a session and sets an HttpOnly, SameSite=lax cookie, while whoami reads request.scope['session'] and returns 401 when it is empty. Because the middleware already refreshed the TTL, every successful call silently pushes the idle timeout forward. This separation keeps expiry policy in the store, transport in the middleware, and authorization in the routes.
Related snips
payload = {
sub: user.id,
iss: 'https://auth.example.com',
aud: 'codesnips-api',
exp: 15.minutes.from_now.to_i,
iat: Time.now.to_i,
JWT issuance and verification without common footguns
json.array! @posts do |post|
json.cache! ['v1', post], expires_in: 1.hour do
json.id post.id
json.title post.title
json.excerpt post.excerpt
json.published_at post.published_at
Fragment caching for expensive JSON serialization
#!/usr/bin/env bash
set -euo pipefail
export VAULT_ADDR="https://vault.internal:8200"
export VAULT_TOKEN="${VAULT_TOKEN:?missing VAULT_TOKEN}"
Secrets management with environment isolation and Vault
import { randomBytes, createHash } from "crypto";
import jwt from "jsonwebtoken";
import { RefreshTokenStore } from "./store";
const ACCESS_SECRET = process.env.ACCESS_SECRET!;
const ACCESS_TTL = "15m";
JWT access + refresh token rotation (conceptual)
package files
import (
"context"
"time"
Presigned S3 upload URLs (AWS SDK v2)
<%# private stream: turbo signs the serialized record name %>
<%= turbo_stream_from current_user %>
<section class="notifications">
<h1>Notifications</h1>
Turbo Streams + authorization: signed per-user stream name
Share this code
Here's the card — post it anywhere.