"""服务端 Session 管理(MVP: 内存 dict + TTL)。 - Session token: secrets.token_urlsafe(32) - Cookie: HttpOnly, SameSite=Lax, 生产环境 Secure - 默认过期: 7 天 未来切换 Redis: 改 _store 为 Redis 客户端,接口不变。 """ import secrets import time from dataclasses import dataclass from app.core.config import get_settings # 内存存储:{token: SessionEntry} _store: dict[str, "SessionEntry"] = {} SESSION_COOKIE_NAME = "session_id" SESSION_TTL_SECONDS = 7 * 24 * 3600 # 7 天 @dataclass class SessionEntry: user_id: str created_at: float last_accessed: float def create_session(user_id: str) -> str: """创建 session,返回 token(写入 Cookie)。""" _cleanup_expired() token = secrets.token_urlsafe(32) now = time.time() _store[token] = SessionEntry(user_id=user_id, created_at=now, last_accessed=now) return token def get_session_user(token: str) -> str | None: """根据 token 获取 user_id;不存在或过期返回 None。""" entry = _store.get(token) if entry is None: return None now = time.time() if now - entry.last_accessed > SESSION_TTL_SECONDS: _store.pop(token, None) return None entry.last_accessed = now return entry.user_id def delete_session(token: str) -> None: _store.pop(token, None) def _cleanup_expired() -> None: """惰性清理过期 session。""" now = time.time() expired = [t for t, e in _store.items() if now - e.last_accessed > SESSION_TTL_SECONDS] for t in expired: _store.pop(t, None) def get_cookie_params() -> dict: """构建 Set-Cookie 参数。""" settings = get_settings() return { "key": SESSION_COOKIE_NAME, "httponly": True, "samesite": "lax", "secure": settings.is_production, "max_age": SESSION_TTL_SECONDS, "path": "/", }