FastAPI开发环境调用Firestore .stream()后接口挂起问题

我正在构建一个使用Firestore作为数据库的FastAPI后端。生产环境（Docker + Gunicorn）运行一切正常，但在开发环境（Docker + Uvicorn）中，调用/login/端点执行Firestore的.stream()查询后，服务会无限挂起，无法处理后续请求，必须终止容器才能恢复。

相关代码 #

@router.post("/login/", tags=["users"])
async def login(user: UserLogin):
    try:
        users_ref = db.collection("users")
        query = users_ref.where("email", "==", user.email).limit(1).stream()
        user_doc = next(query, None)

        if user_doc is None:
            raise HTTPException(status_code=401, detail="Invalid email or password")

        user_data = user_doc.to_dict()
        hashed_password = user_data.get("hashed_password")

        if not bcrypt.checkpw(user.password.encode(), hashed_password.encode()):
            raise HTTPException(status_code=401, detail="Invalid email or password")

        token_data = {
            "sub": user_doc.id,
            "email": user_data["email"],
        }

        access_token = create_access_token(token_data)
        refresh_token = create_refresh_token(token_data)

        return {
            "access_token": access_token,
            "refresh_token": refresh_token,
            "token_type": "bearer"
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Unexpected error: {e}")

已确认事项 #

• 容器中.env文件加载正常
• Firestore凭证（GOOGLE_APPLICATION_CREDENTIALS）有效
• /hello/等其他端点运行正常
• 各环境代码和依赖一致
• 仅开发环境调用/login/后出现问题

重要提示 #

若通过本地venv运行uvicorn app.main:app --reload则无此问题。

疑问 #

1. .stream()是否为阻塞操作，不应在async def函数中调用？
2. 如何解决该问题？
3. 为何仅在容器化开发环境中出现？

Dockerfile内容 #

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

1. .stream()是否为阻塞操作，不应在async def函数中调用？ #

是的，Firestore Python SDK的同步方法（包括.stream()）都是阻塞IO操作。在FastAPI的async def端点中直接调用这类操作会占用整个事件循环，导致服务无法处理后续请求，这就是挂起的核心原因。

2. 如何解决该问题？ #

有两种可靠的解决方式：

方式一：将端点改为同步函数

把async def login改成def login，FastAPI会自动将同步函数放到线程池中执行，不会阻塞事件循环：

@router.post("/login/", tags=["users"])
def login(user: UserLogin):
    try:
        users_ref = db.collection("users")
        query = users_ref.where("email", "==", user.email).limit(1).stream()
        user_doc = next(query, None)

        if user_doc is None:
            raise HTTPException(status_code=401, detail="Invalid email or password")

        user_data = user_doc.to_dict()
        hashed_password = user_data.get("hashed_password")

        if not bcrypt.checkpw(user.password.encode(), hashed_password.encode()):
            raise HTTPException(status_code=401, detail="Invalid email or password")

        token_data = {
            "sub": user_doc.id,
            "email": user_data["email"],
        }

        access_token = create_access_token(token_data)
        refresh_token = create_refresh_token(token_data)

        return {
            "access_token": access_token,
            "refresh_token": refresh_token,
            "token_type": "bearer"
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Unexpected error: {e}")

方式二：用线程池异步执行阻塞操作

如果需要保持端点为async def，可以用asyncio.to_thread将阻塞的Firestore调用包装起来：

import asyncio

@router.post("/login/", tags=["users"])
async def login(user: UserLogin):
    try:
        def get_user_by_email():
            users_ref = db.collection("users")
            query = users_ref.where("email", "==", user.email).limit(1).stream()
            return next(query, None)
        
        # 将阻塞操作放到线程池执行，不占用事件循环
        user_doc = await asyncio.to_thread(get_user_by_email)

        if user_doc is None:
            raise HTTPException(status_code=401, detail="Invalid email or password")
        
        # 剩余代码逻辑不变
        user_data = user_doc.to_dict()
        hashed_password = user_data.get("hashed_password")

        if not bcrypt.checkpw(user.password.encode(), hashed_password.encode()):
            raise HTTPException(status_code=401, detail="Invalid email or password")

        token_data = {
            "sub": user_doc.id,
            "email": user_data["email"],
        }

        access_token = create_access_token(token_data)
        refresh_token = create_refresh_token(token_data)

        return {
            "access_token": access_token,
            "refresh_token": refresh_token,
            "token_type": "bearer"
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Unexpected error: {e}")

另外，推荐用.get()替代.stream()+next的写法，更简洁：

# 替代写法，同样是阻塞操作，需用上述方式包装
user_doc = users_ref.where("email", "==", user.email).limit(1).get()
if not user_doc.exists:
    raise HTTPException(status_code=401, detail="Invalid email or password")

3. 为何仅在容器化开发环境中出现？ #

• 本地venv运行uvicorn --reload时，Uvicorn默认启用多线程模式，阻塞操作会被分配到单独线程，不会完全卡住事件循环。
• 容器化开发环境中，你的Dockerfile里Uvicorn默认使用单线程单worker模式，阻塞操作直接占用唯一的事件循环线程，导致服务彻底无法处理后续请求。
• 生产环境用Gunicorn+Uvicorn时，Gunicorn会启动多个worker进程，每个worker有独立的事件循环，单个阻塞操作只会影响一个worker，其他worker仍能处理请求，因此不会出现全局挂起。

内容的提问来源于stack exchange，提问作者Javier Martín Pizarro