Heroku部署FastAPI+Celery遇Redis NOAUTH认证错误求助

在Heroku部署的架构包含FastAPI API dyno、Celery Beat dyno和Celery Worker dyno，使用Redis Enterprise Cloud、Stackhero for RabbitMQ、Heroku Postgres作为附加组件。目前FastAPI和Celery API可正常使用Redis，但向Worker下发任务时出现错误：

asyncio_redis.exceptions.ErrorReply: NOAUTH Authentication required

相关代码文件 #

配置文件（project/config.py） #

import os
import pathlib
from functools import lru_cache

from kombu import Queue


def route_task(name, args, kwargs, options, task=None, **kw):
    if ":" in name:
        queue, _ = name.split(":")
        return {"queue": queue}
    return {"queue": "default"}


class BaseConfig:
    BASE_DIR: pathlib.Path = pathlib.Path(__file__).parent.parent
    UPLOADS_DEFAULT_DEST: str = str(BASE_DIR / "upload")

    DATABASE_URL: str = os.environ.get("DATABASE_URL", f"sqlite:///{BASE_DIR}/db.sqlite3")
    DATABASE_CONNECT_DICT: dict = {}

    CELERY_BROKER_URL: str = os.environ.get("CELERY_BROKER_URL", "redis://127.0.0.1:6379/0")
    CELERY_RESULT_BACKEND: str = os.environ.get("CELERY_RESULT_BACKEND", "redis://127.0.0.1:6379/0")

    WS_MESSAGE_QUEUE: str = os.environ.get("WS_MESSAGE_QUEUE", "redis://127.0.0.1:6379/0")

    CELERY_BEAT_SCHEDULE: dict = {
        # "task-schedule-work": {
        #     "task": "task_schedule_work",
        #     "schedule": 5.0,  # five seconds
        # },
    }

    CELERY_TASK_DEFAULT_QUEUE: str = "default"

    # Force all queues to be explicitly listed in `CELERY_TASK_QUEUES` to help prevent typos
    CELERY_TASK_CREATE_MISSING_QUEUES: bool = False

    CELERY_TASK_QUEUES: list = (
        # need to define default queue here or exception would be raised
        Queue("default"),

        Queue("high_priority"),
        Queue("low_priority"),
    )

    CELERY_TASK_ROUTES = (route_task,)


class DevelopmentConfig(BaseConfig):
    pass


class ProductionConfig(BaseConfig):
    CELERY_BROKER_URL: str = os.environ.get("STACKHERO_RABBITMQ_AMQP_URL_TLS")
    CELERY_RESULT_BACKEND: str = os.environ.get("REDISCLOUD_URL")
    WS_MESSAGE_QUEUE: str = os.environ.get("REDISCLOUD_URL")
    


class TestingConfig(BaseConfig):
    # https://fastapi.tiangolo.com/advanced/testing-database/
    DATABASE_URL: str = "sqlite:///./test.db"
    DATABASE_CONNECT_DICT: dict = {"check_same_thread": False}


@lru_cache()
def get_settings():
    config_cls_dict = {
        "development": DevelopmentConfig,
        "production": ProductionConfig,
        "testing": TestingConfig
    }

    config_name = os.environ.get("FASTAPI_CONFIG", "development")
    config_cls = config_cls_dict[config_name]
    return config_cls()


settings = get_settings()

Celery工具类（project/celery_utils.py） #

import functools

from celery import current_app as current_celery_app, shared_task
from celery.result import AsyncResult
from celery.utils.time import get_exponential_backoff_interval

from project.config import settings


def create_celery():
    celery_app = current_celery_app
    celery_app.config_from_object(settings, namespace="CELERY")

    return celery_app


def get_task_info(task_id):
    """
    return task info according to the task_id
    """
    task = AsyncResult(task_id)
    state = task.state

    if state == "FAILURE":
        error = str(task.result)
        response = {
            "state": task.state,
            "error": error,
        }
    else:
        response = {
            "state": task.state,
        }
    return response


class custom_celery_task:

    EXCEPTION_BLOCK_LIST = (
        IndexError,
        KeyError,
        TypeError,
        UnicodeDecodeError,
        ValueError,
    )

    def __init__(self, *args, **kwargs):
        self.task_args = args
        self.task_kwargs = kwargs

    def __call__(self, func):
        @functools.wraps(func)
        def wrapper_func(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except self.EXCEPTION_BLOCK_LIST:
                # do not retry for those exceptions
                raise
            except Exception as e:
                # here we add Exponential Backoff just like Celery
                countdown = self._get_retry_countdown(task_func)
                raise task_func.retry(exc=e, countdown=countdown)

        task_func = shared_task(*self.task_args, **self.task_kwargs)(wrapper_func)
        return task_func

    def _get_retry_countdown(self, task_func):
        retry_backoff = int(
            self.task_kwargs.get("retry_backoff", True)
        )
        retry_backoff_max = int(
            self.task_kwargs.get("retry_backoff_max", 600)
        )
        retry_jitter = self.task_kwargs.get(
            "retry_jitter", True
        )

        countdown = get_exponential_backoff_interval(
            factor=retry_backoff,
            retries=task_func.request.retries,
            maximum=retry_backoff_max,
            full_jitter=retry_jitter
        )

        return countdown

应用初始化文件（project/init.py） #

from broadcaster import Broadcast
from fastapi import FastAPI

from project.config import settings

broadcast = Broadcast(settings.WS_MESSAGE_QUEUE)


def create_app() -> FastAPI:
    app = FastAPI()

    from project.logging import configure_logging
    configure_logging()

    # do this before loading routes
    from project.celery_utils import create_celery
    app.celery_app = create_celery()

    from project.users import users_router
    app.include_router(users_router)

    from project.tdd import tdd_router
    app.include_router(tdd_router)

    from project.ws import ws_router
    app.include_router(ws_router)

    from project.ws.views import register_socketio_app
    register_socketio_app(app)

    @app.on_event("startup")
    async def startup_event():
        await broadcast.connect()

    @app.on_event("shutdown")
    async def shutdown_event():
        await broadcast.disconnect()

    @app.get("/")
    async def root():
        return {"message": "Hello World"}

    return app

原因分析与解决方法 #

可能原因 #

1. Worker环境变量配置异常：Worker dyno未正确设置FASTAPI_CONFIG=production，导致加载了BaseConfig中默认的无密码Redis地址；或者REDISCLOUD_URL变量未在Worker dyno中配置/配置错误。
2. Redis URL解析问题：Celery或异步Redis客户端（如broadcaster使用的asyncio_redis）对REDISCLOUD_URL的解析存在异常，无法正确提取认证信息。
3. 任务内部Redis配置错误：任务代码中手动初始化Redis客户端时，未使用settings中的URL，而是硬编码了无密码的本地Redis地址。

解决步骤 #

1. 验证Worker环境变量：

   • 在Worker启动命令前添加环境变量打印，比如echo "FASTAPI_CONFIG=$FASTAPI_CONFIG REDISCLOUD_URL=$REDISCLOUD_URL"，确认变量值与API dyno一致。
   • 确保Heroku的环境变量已正确同步到所有dyno，可通过heroku config --app <your-app-name>查看全局配置。

2. 测试Redis URL有效性：

   • 使用Redis CLI直接连接REDISCLOUD_URL：redis-cli -u $REDISCLOUD_URL，执行PING命令验证认证是否正常。

3. 强制Celery配置生效：
   修改create_celery函数，显式设置result backend，避免namespace加载的潜在问题：

   def create_celery():
       celery_app = current_celery_app
       celery_app.config_from_object(settings, namespace="CELERY")
       # 显式指定结果后端，确保生产环境配置生效
       celery_app.conf.result_backend = settings.CELERY_RESULT_BACKEND
       return celery_app
   

4. 检查任务代码中的Redis使用：
   确认所有任务中操作Redis的逻辑，均使用settings.WS_MESSAGE_QUEUE或settings.CELERY_RESULT_BACKEND作为连接URL，禁止硬编码地址。

5. 验证Worker的Redis连接：
   在Worker启动时添加Redis连接测试逻辑，比如在celery启动脚本中加入：

   import redis
   from project.config import settings
   
   r = redis.from_url(settings.CELERY_RESULT_BACKEND)
   print("Redis connection test:", r.ping())
   

   确认Worker能正常连接Redis。

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