如何在Django Celery定时任务中获取request.user？

首先明确核心问题：Celery Beat触发的定时任务是后台自动执行的，完全脱离HTTP请求生命周期，不存在request对象，所以你之前尝试通过request获取用户的思路从根本上走不通，和中间件也没关系——中间件只处理HTTP请求链路的上下文，管不到定时任务。

下面是具体可行的方案，根据你的任务目标选择：

1. 任务目标：针对所有用户执行操作 #

直接遍历User模型，逐个处理用户逻辑：

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

@shared_task
def run_everyday():
    # 遍历所有活跃用户（可根据需求添加过滤条件）
    for user in User.objects.filter(is_active=True).iterator():
        # 在这里执行针对用户的逻辑，比如发送日报邮件、更新用户数据
        print(f"处理用户：{user.username}")
    return 0

• 效率优化：用iterator()避免一次性把所有用户数据加载到内存，适合用户量较大的场景。

2. 任务目标：针对特定用户执行操作 #

如果任务只需要处理固定用户（比如管理员、某个特定账号），直接通过User模型查询获取：

@shared_task
def run_everyday():
    try:
        # 按用户名或ID查询目标用户
        user = User.objects.get(username='admin')
        # 执行该用户的专属逻辑
        print(f"处理管理员用户：{user.username}")
    except User.DoesNotExist:
        # 处理用户不存在的异常
        print("目标用户不存在")
    return 0

• 效率：单用户查询是数据库快速查询，效率很高，没有性能问题。

3. 任务目标：和用户操作关联的定时任务 #

如果你的任务是用户触发后延迟执行（比如用户点击后24小时执行），而非每日固定时间跑的全局任务，那应该在用户请求时把用户ID传入任务，而非用Beat调度：

# 视图函数中，用户发起请求时调用
def user_trigger_task(request):
    # 获取当前请求用户的ID
    user_id = request.user.id
    # 设置24小时后执行任务，传入用户ID
    run_for_user.apply_async(args=[user_id], countdown=86400)
    return HttpResponse("任务已安排")

# Celery任务函数
@shared_task
def run_for_user(user_id):
    try:
        user = User.objects.get(id=user_id)
        # 执行针对该用户的逻辑
        print(f"处理用户：{user.username}")
    except User.DoesNotExist:
        print("用户不存在，任务终止")
    return 0

关于中间件的疑问 #

不需要创建中间件，中间件仅作用于HTTP请求的处理流程，定时任务不在这个流程里，无法通过中间件获取任何请求上下文，完全是无效方案。

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