如何为Celery任务分配用户以追踪模型CRUD操作执行者？

我正规划一个项目，将使用Celery任务对模型执行CRUD操作。此外，我已配置Signals监听这些CRUD操作，希望专门记录是Celery任务（或Celery任务用户）执行了该操作。请问如何为Celery Worker分配用户，以便在其与模型交互时进行追踪？

对应的信号代码：

@receiver(pre_save, sender=Location)
def location_pre_save(sender, instance, update_fields=None, **kwargs):
    try:
        old_location = Location.objects.get(id=instance.id)
        # Log Celery user has updated this Location
    except Location.DoesNotExist:
        # Log Celery user has created this Location

解决方案 #

方法1：创建Celery专用系统用户+线程本地存储传递 #

1. 创建专用系统用户
   先在数据库中创建一个标记为系统用户的账号，用于所有Celery任务的操作身份：

# 示例：在Django Shell中创建
from django.contrib.auth.models import User
# 创建时可根据需求设置权限，比如设为非staff用户
User.objects.create_user(
    username='celery_worker',
    email='celery_system@yourdomain.com',
    password='your_secure_password',
    is_active=True,
    is_staff=False
)

2. 线程本地存储传递用户
   利用Python的线程本地存储，在Celery任务中临时绑定用户身份：

import threading
# 定义全局线程本地存储容器
local_store = threading.local()

# Celery任务示例
from celery import shared_task
from django.contrib.auth.models import User
from .models import Location

@shared_task
def update_location_task(location_id, new_data):
    # 获取Celery专用用户
    celery_user = User.objects.get(username='celery_worker')
    # 将用户存入线程本地存储
    local_store.current_user = celery_user
    
    # 执行模型操作
    location = Location.objects.get(id=location_id)
    for k, v in new_data.items():
        setattr(location, k, v)
    location.save()
    
    # 清理存储，避免线程复用残留数据
    del local_store.current_user

3. 信号中读取用户并记录
   修改信号函数，从线程存储中获取当前用户：

import threading
local_store = threading.local()

@receiver(pre_save, sender=Location)
def location_pre_save(sender, instance, update_fields=None, **kwargs):
    current_user = getattr(local_store, 'current_user', None)
    try:
        old_location = Location.objects.get(id=instance.id)
        if current_user and current_user.username == 'celery_worker':
            # 记录Celery用户更新操作，这里替换为你的日志逻辑
            print(f"Celery用户 {current_user.username} 更新了Location[{instance.id}]")
    except Location.DoesNotExist:
        if current_user and current_user.username == 'celery_worker':
            # 记录Celery用户创建操作
            print(f"Celery用户 {current_user.username} 创建了Location[{instance.id}]")

方法2：模型新增操作人字段（更直观） #

直接在模型中添加关联用户的字段，Celery任务执行时直接赋值，信号中直接读取：

1. 修改模型

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

class Location(models.Model):
    # 原有字段...
    created_by = models.ForeignKey(
        User, 
        on_delete=models.SET_NULL, 
        null=True, 
        related_name='created_locations'
    )
    updated_by = models.ForeignKey(
        User, 
        on_delete=models.SET_NULL, 
        null=True, 
        related_name='updated_locations'
    )

执行makemigrations和migrate命令更新数据库结构。

2. Celery任务中直接赋值

from celery import shared_task
from django.contrib.auth.models import User
from .models import Location

@shared_task
def create_location_task(location_data):
    celery_user = User.objects.get(username='celery_worker')
    Location.objects.create(**location_data, created_by=celery_user)

@shared_task
def update_location_task(location_id, new_data):
    celery_user = User.objects.get(username='celery_worker')
    Location.objects.filter(id=location_id).update(
        **new_data, 
        updated_by=celery_user
    )

3. 信号中直接读取字段记录

@receiver(pre_save, sender=Location)
def location_pre_save(sender, instance, update_fields=None, **kwargs):
    try:
        old_location = Location.objects.get(id=instance.id)
        if instance.updated_by and instance.updated_by.username == 'celery_worker':
            # 记录Celery更新操作
            print(f"Celery用户 {instance.updated_by.username} 更新了Location[{instance.id}]")
    except Location.DoesNotExist:
        if instance.created_by and instance.created_by.username == 'celery_worker':
            # 记录Celery创建操作
            print(f"Celery用户 {instance.created_by.username} 创建了Location[{instance.id}]")

注意事项 #

• 推荐使用方法2，模型字段存储操作人不仅能满足信号追踪需求，还能直接通过模型查询所有Celery执行的操作，更便于后续排查和统计。
• 使用线程本地存储时，要注意Celery Worker的多进程/多线程特性，线程存储不会跨进程共享，无需担心数据冲突，但任务结束后建议清理存储避免线程复用残留。
• 确保celery_worker用户拥有足够的权限执行对应模型的CRUD操作，避免权限报错。

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