如何修改Django中间件实现每次请求更新用户last_activity字段？

1. 给User模型添加last_activity字段 #

如果使用自定义用户模型，修改models.py：

from django.contrib.auth.models import AbstractUser
from django.db import models
from django.utils import timezone

class CustomUser(AbstractUser):
    last_activity = models.DateTimeField(default=timezone.now)

然后在settings.py中指定自定义用户模型：

AUTH_USER_MODEL = 'your_app_name.CustomUser'

若不想替换默认User模型，也可创建关联的Profile模型，通过OneToOneField关联User并添加last_activity字段。

2. 编写全局更新last_activity的中间件 #

修改中间件逻辑，让它在用户每次发起请求时更新字段，而非仅登录时。创建middleware.py文件：

from django.utils import timezone
from django.contrib.auth import get_user_model

User = get_user_model()

class UpdateLastActivityMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # 仅对已认证用户更新last_activity
        if request.user.is_authenticated:
            # 使用filter+update提升性能，避免全字段保存
            User.objects.filter(pk=request.user.pk).update(last_activity=timezone.now())
        
        # 继续处理请求并返回响应
        response = self.get_response(request)
        return response

在settings.py的MIDDLEWARE列表中添加该中间件，必须放在AuthenticationMiddleware之后：

MIDDLEWARE = [
    # 其他系统中间件...
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'your_app_name.middleware.UpdateLastActivityMiddleware',
    # 其他自定义中间件...
]

3. 获取在线用户（基于5分钟阈值） #

编写工具函数筛选最近5分钟有活动的用户：

from django.utils import timezone
from django.contrib.auth import get_user_model

User = get_user_model()

def get_online_users():
    five_minutes_ago = timezone.now() - timezone.timedelta(minutes=5)
    return User.objects.filter(last_activity__gte=five_minutes_ago)

可在视图或模板中调用该函数，展示在线用户列表。

注意事项 #

• 性能优化：使用filter(pk=...).update(...)而非request.user.save()，前者仅更新指定字段，减少数据库开销，也不会触发不必要的模型信号。
• 桌面应用适配：若桌面应用使用Token认证（如JWT），需调整中间件识别Token用户：

from rest_framework_simplejwt.authentication import JWTAuthentication
from django.utils import timezone
from django.contrib.auth import get_user_model

User = get_user_model()

class UpdateLastActivityMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # 处理Session认证用户
        if request.user.is_authenticated:
            User.objects.filter(pk=request.user.pk).update(last_activity=timezone.now())
        else:
            # 处理JWT Token认证的桌面应用用户
            try:
                jwt_auth = JWTAuthentication()
                user, _ = jwt_auth.authenticate(request)
                if user:
                    User.objects.filter(pk=user.pk).update(last_activity=timezone.now())
            except Exception:
                # 忽略Token无效或不存在的情况
                pass
        
        response = self.get_response(request)
        return response

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