基于视图kwargs编写Django SessionMiddleware 实现特定URL不保存会话

问题根因 #

原有代码失效的核心问题有3个：

• process_request阶段直接将request.session赋值为普通字典，丢失了SessionStore自带的accessed/modified/is_empty()等核心属性和方法，会直接触发AuthenticationMiddleware的断言错误或属性异常
• process_view没有从路由参数中提取client_url做状态标记，也没有按需初始化对应数据库的会话实例
• process_response没有增加会话保存的前置判断，不管URL是否符合要求都会执行会话保存逻辑

核心修改逻辑 #

1. 新增request.should_save_session标记位，默认值为False，仅当URL携带client_url参数时才设为True
2. process_request阶段统一初始化空的会话实例，避免后续中间件报错，不做业务逻辑判断
3. process_view阶段从路由关键字参数中提取client_url，如果存在则重新初始化对应数据库的会话实例，同时打开保存开关
4. process_response阶段优先判断保存开关，开关关闭时直接跳过所有会话保存、Cookie设置逻辑

完整修改后代码 #

import time
from importlib import import_module

from django.conf import settings
from django.contrib.sessions.backends.base import UpdateError
from django.contrib.sessions.exceptions import SessionInterrupted
from django.utils.cache import patch_vary_headers
from django.utils.deprecation import MiddlewareMixin
from django.utils.functional import SimpleLazyObject
from django.utils.http import http_date


class SessionMiddleware(MiddlewareMixin):
    def __init__(self, get_response=None):
        super().__init__(get_response)
        engine = import_module(settings.SESSION_ENGINE)
        self.SessionStore = engine.SessionStore

    def process_request(self, request):
        session_key = request.COOKIES.get(settings.SESSION_COOKIE_NAME)
        # 先初始化默认会话实例，避免鉴权中间件报错，默认不保存
        request.session = self.SessionStore(session_key)
        # 新增保存标记位，默认关闭
        request.should_save_session = False

    def process_view(self, request, callback, callback_args, callback_kwargs):
        # 从路由关键字参数中提取client_url
        client_url = callback_kwargs.get('client_url')
        if not client_url:
            return
        # 匹配到client_url参数，打开保存开关
        request.should_save_session = True
        # --- 此处添加你的多数据库会话初始化逻辑 ---
        # 示例：如果不同client_url对应不同数据库，可指定会话的using属性
        # request.session.using = get_db_alias_by_client_url(client_url)
        # 也可以重新初始化对应数据库的SessionStore实例
        # ----------------------------------------

    def process_response(self, request, response):
        # 优先判断保存开关，关闭时直接返回响应
        if not getattr(request, 'should_save_session', False):
            return response
        try:
            accessed = request.session.accessed
            modified = request.session.modified
            empty = request.session.is_empty()
        except AttributeError:
            return response
        if settings.SESSION_COOKIE_NAME in request.COOKIES and empty:
            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:
            if accessed:
                patch_vary_headers(response, ('Cookie',))
            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:
                        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."
                        )
                    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

from django.contrib import auth

def get_user(request):
    if not hasattr(request, '_cached_user'):
        request._cached_user = auth.get_user(request)
    return request._cached_user

class AuthenticationMiddleware(MiddlewareMixin):
    def process_request(self, request):
        assert hasattr(request, 'session'), (
            "The Django authentication middleware requires session middleware "
            "to be installed. Edit your MIDDLEWARE setting to insert "
            "'django.contrib.sessions.middleware.SessionMiddleware' before "
            "'django.contrib.auth.middleware.AuthenticationMiddleware'."
        )
        request.user = SimpleLazyObject(lambda: get_user(request))

注意事项 #

• 确保你的URLconf中对应的路由确实将client_url作为关键字参数传递，否则process_view无法提取到参数
• 多数据库场景下，需要自行实现get_db_alias_by_client_url逻辑，将会话绑定到对应的数据库
• 替换原有Django默认的SessionMiddleware为你自定义的这个中间件即可生效

内容的提问来源于stack exchange，提问作者ViaTech