Django pytest异步任务报错：用户对象PK无效，对象不存在

这个问题本质是测试环境下Celery异步任务与主测试进程的数据库隔离导致的，非测试环境正常是因为生产环境中数据库操作是持久化的，而测试环境的事务机制会让异步进程看不到未提交的数据。以下是具体排查点和解决方法：

1. 测试事务的隔离问题 #

pytest-django默认会为每个测试用例开启数据库事务，测试结束后自动回滚。但Celery任务是独立进程，无法访问主测试进程中事务内未提交的数据——哪怕你在视图里创建了用户，只要事务没提交，Celery进程就查不到这个用户。

解决方法：

• 给测试用例添加@pytest.mark.django_db(transaction=False)，禁用事务回滚，让数据直接提交到测试数据库，Celery进程就能读取到。
• 或者使用django_db_reset_sequences fixture，确保测试数据能被其他进程访问。

示例：

import pytest
from celery.result import AsyncResult

@pytest.mark.django_db(transaction=False)
def test_create_custom_object(authenticated_api_client):
    # 发送POST请求触发异步任务
    response = authenticated_api_client.post("/api/custom-objects/", data={"name": "test"})
    task_id = response.data["task_id"]
    
    # 等待任务执行完成
    result = AsyncResult(task_id).get(timeout=10)
    
    # 验证结果
    assert result["status"] == "success"

2. 确保Celery使用测试数据库 #

测试环境中要强制Celery任务连接到测试数据库，而非默认的生产数据库。可以通过pytest fixture覆盖Celery的配置：

import pytest
from django.conf import settings
from celery import current_app

@pytest.fixture(autouse=True)
def configure_celery_test_db():
    # 让Celery使用Django测试数据库的配置
    db_config = settings.DATABASES["default"]
    current_app.conf.database_engine_options = db_config
    yield

3. 同步执行Celery任务（快速排查用） #

如果只是想验证业务逻辑，不想处理异步进程的隔离问题，可以在测试中强制Celery任务同步执行：

@pytest.fixture(autouse=True)
def celery_eager_mode(settings):
    settings.CELERY_TASK_ALWAYS_EAGER = True
    settings.CELERY_TASK_EAGER_PROPAGATES = True

这样所有Celery任务会在主进程中同步执行，完全避免跨进程的数据库隔离问题，适合快速验证业务逻辑是否正确。

4. 检查用户ID的传递与解析 #

确认任务中接收的用户ID格式正确：

• 如果用户主键是UUID，确保传递的是字符串格式的UUID，任务中直接用User.objects.get(pk=user_id)即可（Django支持字符串UUID查询主键）。
• 避免在任务中使用缓存的用户数据，必须重新从数据库查询。

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