Django 提供了一个强大的会话（Session）系统，用于在多个请求之间存储和检索用户特定的数据。
1、系统结构
1.1 代码结构
django.contrib.sessions ├── middleware.py │ └── SessionMiddleware (处理请求/响应周期) ├── backends/ │ ├── base.py │ │ └── SessionBase (抽象基类) │ ├── db.py │ │ └── SessionStore (数据库后端) │ ├── cache.py │ │ └── CacheSession (缓存后端) │ └── signed_cookies.py │ └── SignedCookieSession (Cookie后端) └── models.py └── Session (数据库模型) 1.2 工作原理
- 当用户首次访问时，Django 创建一个唯一的 session ID
- 这个 ID 通常通过 cookie 发送到客户端
- 后续请求中，客户端会携带这个 ID，Django 通过它找到对应的 session 数据
- Session 数据存储在服务器端（数据库、缓存等）
2、源码分析
2.1 SessionMiddleware
process_request
: 处理请求，获取session_key并获取session赋值给requestprocess_response
：处理响应，保存 session 并设置 Cookie
class SessionMiddleware(MiddlewareMixin): def __init__(self, get_response): super().__init__(get_response) # 导入 SessionStore 类 engine = import_module(settings.SESSION_ENGINE) self.SessionStore = engine.SessionStore def process_request(self, request): # 从 Cookie 中获取 session_key session_key = request.COOKIES.get(settings.SESSION_COOKIE_NAME) # 创建 SessionStore 实例并赋值给 request.session request.session = self.SessionStore(session_key) def process_response(self, request, response): """ 处理响应，保存 session 并设置 Cookie """ try: accessed = request.session.accessed modified = request.session.modified empty = request.session.is_empty() except AttributeError: # 如果没有 session 属性，直接返回响应 return response if settings.SESSION_COOKIE_NAME in request.COOKIES and empty: # 如果 session 为空但 Cookie 存在，删除 Cookie response.delete_cookie( settings.SESSION_COOKIE_NAME, path=settings.SESSION_COOKIE_PATH, domain=settings.SESSION_COOKIE_DOMAIN, samesite=settings.SESSION_COOKIE_SAMESITE, ) patch_vary_headers(response, ("Cookie",)) else: # 如果 session 被访问过，设置 Vary: Cookie 头 if accessed: patch_vary_headers(response, ("Cookie",)) # 如果 session 被修改或设置了 SESSION_SAVE_EVERY_REQUEST，并且非空，则保存 if (modified or settings.SESSION_SAVE_EVERY_REQUEST) and not empty: if request.session.get_expire_at_browser_close(): max_age = None expires = None else: max_age = request.session.get_expiry_age() expires_time = time.time() + max_age expires = http_date(expires_time) if response.status_code < 500: try: # 保存 session request.session.save() except UpdateError: raise SessionInterrupted( "The request's session was deleted before the " "request completed. The user may have logged " "out in a concurrent request, for example." ) # 设置 Cookie response.set_cookie( settings.SESSION_COOKIE_NAME, request.session.session_key, max_age=max_age, expires=expires, domain=settings.SESSION_COOKIE_DOMAIN, path=settings.SESSION_COOKIE_PATH, secure=settings.SESSION_COOKIE_SECURE or None, httponly=settings.SESSION_COOKIE_HTTPONLY or None, samesite=settings.SESSION_COOKIE_SAMESITE, ) return response 2.2 SessionBase 抽象基类关键方法
class SessionBase: """ Session 基类，定义了所有 Session 后端的通用接口 """ def __init__(self, session_key=None): self._session_key = session_key self.accessed = False self.modified = False self.serializer = import_string(settings.SESSION_SERIALIZER) def __getitem__(self, key): return self._session[key] def __setitem__(self, key, value): self._session[key] = value self.modified = True def save(self, must_create=False): """ 必须由子类实现的具体保存逻辑 """ raise NotImplementedError( "subclasses of SessionBase must provide a save() method" ) def load(self): """ 必须由子类实现的数据加载逻辑 """ raise NotImplementedError( "subclasses of SessionBase must provide a load() method" ) def is_empty(self): # 返回 session 是否为空 try: return not self._session_key and not self._session_cache except AttributeError: return True def _get_session(self, no_load=False): """ 获取 session 数据，如果没有加载则加载 """ self.accessed = True try: return self._session_cache except AttributeError: if self.session_key is None or no_load: self._session_cache = {} else: self._session_cache = self.load() return self._session_cache def get_expiry_age(self, **kwargs): """ 获取 session 过期时间（秒） """ try: modification = kwargs["modification"] except KeyError: modification = timezone.now() try: expiry = kwargs["expiry"] except KeyError: expiry = self.get("_session_expiry") if not expiry: return self.get_session_cookie_age() if not isinstance(expiry, (datetime, str)): return expiry if isinstance(expiry, str): expiry = datetime.fromisoformat(expiry) delta = expiry - modification return delta.days * 86400 + delta.seconds def get_expiry_date(self, **kwargs): """ 获取 session 过期日期 """ try: modification = kwargs["modification"] except KeyError: modification = timezone.now() try: expiry = kwargs["expiry"] except KeyError: expiry = self.get("_session_expiry") if isinstance(expiry, datetime): return expiry elif isinstance(expiry, str): return datetime.fromisoformat(expiry) expiry = expiry or self.get_session_cookie_age() return modification + timedelta(seconds=expiry) def get_expire_at_browser_close(self): """ 获取是否在浏览器关闭时过期 """ if (expiry := self.get("_session_expiry")) is None: return settings.SESSION_EXPIRE_AT_BROWSER_CLOSE return expiry == 0 def set_expiry(self, value): """ 设置 session 过期时间 """ if value is None: # 使用全局设置 try: del self["_session_expiry"] except KeyError: pass return if isinstance(value, timedelta): value = timezone.now() + value if isinstance(value, datetime): value = value.isoformat() self["_session_expiry"] = value def flush(self): """ 清空 session 并生成新的 session_key """ self.clear() self.delete() self._session_key = None def cycle_key(self): """ 生成新的 session_key，保持 session 数据 """ data = self._session key = self.session_key self.create() self._session_cache = data if key: self.delete(key) 2.3 数据库Session后端源码
class SessionStore(SessionBase): """ 数据库支持的 session 存储 """ def __init__(self, session_key=None): super().__init__(session_key) @classmethod def get_model_class(cls): # 避免循环导入，获取Session数据库模型定义 from django.contrib.sessions.models import Session return Session @cached_property def model(self): return self.get_model_class() def _get_session_from_db(self): # 从数据库获取session实例 return self.model.objects.get( session_key=self.session_key, expire_date__gt=timezone.now() ) def load(self): # 解码session s = self._get_session_from_db() return self.decode(s.session_data) if s else {} def exists(self, session_key): # 判断session是否存在 return self.model.objects.filter(session_key=session_key).exists() def create(self): while True: self._session_key = self._get_new_session_key() try: # 保存入库 self.save(must_create=True) except CreateError: # Key wasn't unique. Try again. continue self.modified = True return def create_model_instance(self, data): """ """ return self.model( session_key=self._get_or_create_session_key(), session_data=self.encode(data), expire_date=self.get_expiry_date(), ) def save(self, must_create=False): """ 使用事务保存session """ if self.session_key is None: return self.create() data = self._get_session(no_load=must_create) obj = self.create_model_instance(data) using = router.db_for_write(self.model, instance=obj) with transaction.atomic(using=using): obj.save( force_insert=must_create, force_update=not must_create, using=using ) def delete(self, session_key=None): # 删除 self.model.objects.get(session_key=session_key).delete() @classmethod def clear_expired(cls): # 清理过期session cls.get_model_class().objects.filter(expire_date__lt=timezone.now()).delete() 3、配置和使用
3.1 配置
在 settings.py 中配置 Session
# Session 引擎 SESSION_ENGINE = 'django.contrib.sessions.backends.db' # 数据库 # SESSION_ENGINE = 'django.contrib.sessions.backends.cache' # 缓存 # SESSION_ENGINE = 'django.contrib.sessions.backends.cached_db' # 缓存+数据库 # SESSION_ENGINE = 'django.contrib.sessions.backends.file' # 文件 # SESSION_ENGINE = 'django.contrib.sessions.backends.signed_cookies' # 签名Cookie # Session Cookie 名称 SESSION_COOKIE_NAME = 'sessionid' # Session Cookie 过期时间（秒） SESSION_COOKIE_AGE = 1209600 # 2周 # 是否在浏览器关闭时过期 SESSION_EXPIRE_AT_BROWSER_CLOSE = False # 是否每次请求都保存 Session SESSION_SAVE_EVERY_REQUEST = False # Session Cookie 路径 SESSION_COOKIE_PATH = '/' # Session Cookie 域名 SESSION_COOKIE_DOMAIN = None # 是否只通过 HTTPS 传输 SESSION_COOKIE_SECURE = False # 是否阻止 JavaScript 访问 SESSION_COOKIE_HTTPONLY = True # SameSite 属性 SESSION_COOKIE_SAMESITE = 'Lax' # Session 序列化器 SESSION_SERIALIZER = 'django.contrib.sessions.serializers.JSONSerializer' 3.2 基本使用
from django.shortcuts import render, redirect from django.http import HttpResponse def set_session(request): """设置 Session 值""" # 设置简单值 request.session['username'] = 'john' request.session['user_id'] = 123 # 设置复杂数据结构 request.session['user_data'] = { 'name': 'John Doe', 'email': 'john@example.com', 'preferences': { 'theme': 'dark', 'language': 'en' } } # 设置过期时间（300秒后过期） request.session.set_expiry(300) return HttpResponse("Session values set") def get_session(request): """获取 Session 值""" username = request.session.get('username', 'Guest') user_id = request.session.get('user_id') # 获取整个 Session 数据 session_data = dict(request.session.items()) return HttpResponse(f"Username: {username}, User ID: {user_id}") def delete_session(request): """删除 Session 值""" # 删除特定键 if 'username' in request.session: del request.session['username'] # 使用 pop 方法 user_id = request.session.pop('user_id', None) # 清空整个 Session request.session.clear() return HttpResponse("Session values deleted") def flush_session(request): """完全清空 Session 并生成新 Session ID""" request.session.flush() return HttpResponse("Session flushed") 3.3 过期时间
def set_temp_session(request): # 设置临时数据，10分钟后过期 request.session['temp_data'] = '这是临时数据' request.session.set_expiry(600) # 600秒=10分钟 return HttpResponse("临时session数据已设置，10分钟后过期") def set_browser_close_session(request): # 设置浏览器关闭时过期的session request.session['temporary'] = '浏览器关闭后消失' request.session.set_expiry(0) # 0表示浏览器关闭时过期 return HttpResponse("此session将在浏览器关闭时过期") 
这一切，似未曾拥有