如何在Django中监控JWT令牌过期并更新模型字段

JWT本身是无状态令牌，没法主动“监控”过期事件，但可以通过以下几种方式实现你要的需求：

1. 请求拦截时实时处理（最推荐） #

每次用户发起请求验证access token时，若检测到令牌过期，立即更新目标模型字段。可以通过自定义JWT认证后端实现：

自定义JWT认证后端示例（基于djangorestframework-simplejwt） #

from rest_framework_simplejwt.authentication import JWTAuthentication
from rest_framework_simplejwt.exceptions import TokenExpiredError
from django.contrib.auth import get_user_model

User = get_user_model()

class CustomJWTAuthentication(JWTAuthentication):
    def authenticate(self, request):
        try:
            return super().authenticate(request)
        except TokenExpiredError:
            # 从请求中解析令牌及用户ID
            raw_token = self.get_raw_token(
                request.COOKIES.get('access') 
                or request.headers.get('Authorization').split()[1]
            )
            payload = self.get_validated_token(raw_token)
            user_id = payload.get('user_id')
            
            # 更新目标字段，比如将User模型的is_online设为False
            User.objects.filter(id=user_id).update(is_online=False)
            
            # 继续抛出异常，让前端处理过期逻辑
            raise

然后在项目settings.py中配置这个自定义认证类：

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'your_app.authentication.CustomJWTAuthentication',
    ],
}

2. 结合刷新令牌逻辑处理 #

如果你的系统使用refresh token获取新access token，可在刷新视图中添加逻辑：当旧access token过期时更新字段，示例如下：

from rest_framework_simplejwt.views import TokenRefreshView
from rest_framework_simplejwt.exceptions import TokenExpiredError
from rest_framework_simplejwt.tokens import RefreshToken
from django.contrib.auth import get_user_model

User = get_user_model()

class CustomTokenRefreshView(TokenRefreshView):
    def post(self, request, *args, **kwargs):
        try:
            response = super().post(request, *args, **kwargs)
            # 刷新成功说明旧access token已过期，更新字段
            refresh_token = request.data.get('refresh')
            payload = RefreshToken(refresh_token).payload
            User.objects.filter(id=payload['user_id']).update(is_online=False)
            return response
        except TokenExpiredError:
            # 刷新令牌也过期时，同样更新字段
            try:
                refresh_token = request.data.get('refresh')
                payload = RefreshToken(refresh_token).payload
                User.objects.filter(id=payload['user_id']).update(is_online=False)
            except:
                pass
            raise

3. 定时任务批量处理（非实时场景） #

如果不需要实时更新，可使用Celery等定时任务工具，定期扫描并更新已过期令牌对应的字段：

from celery import shared_task
from django.utils import timezone
from django.contrib.auth import get_user_model
from rest_framework_simplejwt.tokens import AccessToken

User = get_user_model()

@shared_task
def update_fields_for_expired_tokens():
    current_time = timezone.now()
    # 需提前在用户模型中保存最近生成的access token
    users = User.objects.exclude(access_token__isnull=True)
    
    for user in users:
        try:
            token = AccessToken(user.access_token)
            if token['exp'] <= current_time.timestamp():
                user.is_online = False
                user.save()
        except:
            # 令牌无效或已过期，直接更新字段
            user.is_online = False
            user.save()

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