Celery 无法与 fastapi 中的 pytest 一起使用

Question

我正在努力测试包含 celery 任务的 fastapi 端点。我想嘲笑芹菜任务。任务如下

send_email.delay(
        user.email,
        "Subject goes here",
        html_content,
    )

这里是实际的模拟功能

@pytest.fixture
def mocked_send_email(mocker):
    mocker.patch('app.views.user.send_email.delay', return_value=None)

def test_create_user(client, mocked_send_email):

    payload = {}

    response = client.post(
        f"{base_url}/user",
        json=payload,
        headers=admin_headers
    )

    mocked_send_email.assert_called_once_with(
        "[email protected]",
        "Subject goes here",
        mock.ANY
    )

这是我处理芹菜的部分

conftest.py

@pytest.fixture(scope="session")
def celery_config():
    print('running celery config')
    return {"broker_url": REDIS, "result_backend": REDIS}


@pytest.fixture(scope='session', autouse=True)
def configure_celery(celery_config):
    from app.celery import celery
    celery.conf.update(celery_config)
    yield
    celery.conf.update({
        'broker_url': 'memory://',
        'result_backend': 'rpc://'
    })

现在当我运行测试时，出现以下错误

usr/local/lib/python3.12/site-packages/kombu/connection.py:476: OperationalError
test_1           | ------------------------------ Captured log call -------------------------------
test_1           | WARNING  kombu.connection:connection.py:669 _info - _info - No hostname was supplied. Reverting to default 'localhost'

我的设置是 dockerized 的，当我不运行 pytest 时，运行不会出现错误。我真的很困惑，考虑到我的设置看起来不错并且在我不运行测试时可以工作，我不确定是什么导致了这个错误。任何帮助将非常感激

更新。打印配置 celery 夹具的语句

Before update:redis://redis-cache:6379/1 redis://redis-cache:6379/1
After update:redis://redis-cache:6379/5 redis://redis-cache:6379/5
After reset: memory:// rpc://

Answer 1

我在使用

@shared_task

装饰器标记 Celery 任务时遇到了问题。根据 Django 的 Celery 文档here，使用

@shared_task

需要额外的配置。由于我使用的是 FastAPI，我不确定如何将此配置合并到我的项目中。

为了解决此问题，我已从使用

@shared_task

切换为直接使用带有

@celery_app.task

的 Celery 应用程序实例。这种方法目前对我有用，但我愿意接受任何能够更好地与 Django 的 Celery 配置保持一致的解决方案。

如果有人对在 FastAPI 项目中正确配置

@shared_task

或更优雅的解决方案有见解，我将不胜感激您的意见。