无request对象时，如何自动更新Django BaseModel的created_by/updated_by？

针对你定义的抽象BaseModel，无法直接获取request对象时，推荐以下几种靠谱的实现方式：

方法1：线程局部存储+信号（最通用） #

这个方法能在全局范围内自动处理，不需要在每次调用时手动传用户：

步骤1：创建线程局部存储工具 #

新建utils/local_storage.py：

import threading

_local = threading.local()

def set_current_user(user):
    _local.user = user

def get_current_user():
    return getattr(_local, 'user', None)

步骤2：编写中间件存储当前用户 #

新建middleware/user_middleware.py：

from django.utils.deprecation import MiddlewareMixin
from utils.local_storage import set_current_user

class CurrentUserMiddleware(MiddlewareMixin):
    def process_request(self, request):
        if request.user.is_authenticated:
            set_current_user(request.user)
        else:
            set_current_user(None)

然后在settings.py的MIDDLEWARE列表末尾添加这个中间件：

MIDDLEWARE = [
    # 其他中间件...
    'middleware.user_middleware.CurrentUserMiddleware',
]

步骤3：用信号自动赋值 #

新建signals/model_signals.py：

from django.db.models.signals import pre_save
from django.dispatch import receiver
from .models import BaseModel
from utils.local_storage import get_current_user

@receiver(pre_save)
def set_created_updated_by(sender, instance, **kwargs):
    # 只处理继承BaseModel的模型
    if issubclass(sender, BaseModel):
        current_user = get_current_user()
        if current_user:
            # 新建对象时设置created_by
            if instance._state.adding:
                instance.created_by = current_user
            # 每次保存都更新updated_by
            instance.updated_by = current_user

记得在apps.py里注册信号：

from django.apps import AppConfig

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app'

    def ready(self):
        import your_app.signals.model_signals

方法2：自定义模型管理器（手动传参，更可控） #

如果不想用全局信号，可以给模型加自定义管理器，要求调用时传入用户：

from django.db import models
from django.contrib.auth import get_user_model

class BaseModelManager(models.Manager):
    def create_with_user(self, user, **kwargs):
        instance = self.model(**kwargs)
        instance.created_by = user
        instance.updated_by = user
        instance.save()
        return instance

    def update_with_user(self, user, instance, **kwargs):
        for key, value in kwargs.items():
            setattr(instance, key, value)
        instance.updated_by = user
        instance.save()
        return instance

# 修改你的BaseModel
class BaseModel(models.Model):
    updated_by = models.ForeignKey(get_user_model(), related_name='+', on_delete=models.RESTRICT, db_column='updated_by')
    updated_at = models.DateTimeField(auto_now=True)
    created_by = models.ForeignKey(get_user_model(), related_name='+', on_delete=models.RESTRICT, db_column='created_by')
    created_at = models.DateTimeField(auto_now_add=True)

    objects = BaseModelManager()

    class Meta:
        abstract = True

调用示例：

# 创建对象
MyModel.objects.create_with_user(user=current_user, name="test")
# 更新对象
obj = MyModel.objects.get(id=1)
MyModel.objects.update_with_user(user=current_user, instance=obj, name="updated test")

方法3：视图层直接赋值（适用于DRF或普通视图） #

如果你的模型操作都在视图里，可以直接在视图层处理：

普通类视图示例： #

from django.views.generic.edit import CreateView, UpdateView
from .models import MyModel

class MyModelCreateView(CreateView):
    model = MyModel
    fields = ['name']

    def form_valid(self, form):
        form.instance.created_by = self.request.user
        form.instance.updated_by = self.request.user
        return super().form_valid(form)

class MyModelUpdateView(UpdateView):
    model = MyModel
    fields = ['name']

    def form_valid(self, form):
        form.instance.updated_by = self.request.user
        return super().form_valid(form)

DRF视图示例： #

from rest_framework.generics import CreateAPIView, UpdateAPIView
from .models import MyModel
from .serializers import MyModelSerializer

class MyModelCreateAPIView(CreateAPIView):
    serializer_class = MyModelSerializer

    def perform_create(self, serializer):
        serializer.save(created_by=self.request.user, updated_by=self.request.user)

class MyModelUpdateAPIView(UpdateAPIView):
    serializer_class = MyModelSerializer
    queryset = MyModel.objects.all()

    def perform_update(self, serializer):
        serializer.save(updated_by=self.request.user)

注意事项 #

• 如果用ASGI异步服务，把threading.local换成asyncio.local，中间件也要改成异步版本
• 信号里要判断用户是否已认证，避免匿名用户赋值报错
• 自定义管理器的方法需要确保调用时都传入用户，适合对权限控制要求严格的场景

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