Django Session
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赋值给request
[*]process_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 response2.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
def __setitem__(self, key, value):
self._session = 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将在浏览器关闭时过期")
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！ 懂技术并乐意极积无私分享的人越来越少。珍惜 用心讨论，共获提升！ 前排留名，哈哈哈 用心讨论，共获提升！ 感谢分享，学习下。
页:
[1]