FastAPI，Pytest TypeError：对象字典不能用于“等待”表达式

Question

当我使用 uvicorn 运行时，我正在使用 FastAPI 编写 API，一切正常，当我想使用 FastAPI TestClient 运行测试时出现错误。

这是错误：

async def get_user_id(conn, user):
   collection = conn.CIA.get_collection("Employees_Info")
   user = await collection.find_one({'name':user},{"_id":1, "name":0, "password":0})
   TypeError: object dict can't be used in 'await' expression
db\db.py:12: TypeError

项目结构：

APP
 |--__init__.py
 |--run.py
 |--main.py
 |--test
    |--test_app.py
 |--routes
    |--router.py
 |--models
    |--models.py
 |--db
    |--db_conn.py
    |--db.py
 |--auth_jwt
    |--jwt_auth.py
 |--auth
    |--auth.py

这是测试的代码，我用的是mongomock，不知道会不会是问题的根源：

import collections
from fastapi.testclient import TestClient
from fastapi import status
from main import app
from mongoengine import connect, disconnect, get_connection
from db.db_conn import db


client = TestClient(app)

connect('mongoenginetest', host='mongomock://localhost', alias='testdb')
db.client = get_connection('testdb')
db.client["CIA"]
db.client["Employees_Info"]
db.client.CIA.Employees_Info.insert_one({"name": "user_Name","password": "week"})

def test_ping():
    response = client.get("/")
    assert response.status_code == status.HTTP_200_OK
    assert response.json() == {"message": "Conectado"}
   
def test_login():
    data = {"username":'user_name', 'password':'week'}
    response = client.post("/login", data=data)
    assert response.headers['Content-Type'] == 'application/json'
    assert response.status_code == status.HTTP_200_OK
    db.client.disconnect()

我尝试根据 FastAPI 文档执行异步测试，但它也不起作用，如果我使用“正常”数据库，则测试可以工作。

路由器.py

@router.post("/login", tags=["user"], response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), 
                                 db: AsyncIOMotorClient = Depends(get_database)):
    authenticate_user_id = await authenticate_user(db, form_data.username, form_data.password)
    if not authenticate_user_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"user_id": str(authenticate_user_id["_id"])})
    return {"access_token": access_token, "token_type": "bearer"}

auth.py

async def authenticate_user(conn:AsyncIOMotorClient, username, password):
    user_id = await verify_user(conn, username)
    if not user_id:
        return False
    
    if not await verify_password(conn, user_id, password):
        return False
    
    return user_id

async def verify_user(conn, user):
    return  await get_user_id(conn,user)

async def verify_password(conn, user_id, password):
    return pbkdf2_sha256.verify(password, await get_password(conn, user_id))

db.py

async def get_user_id(conn, user):
    collection = conn.CIA.get_collection("Employees_Info")
    user = await collection.find_one({'name':user},{"_id":1, "name":0, "password":0})
    print(type(user))
    if user:
        return user

async def get_password(conn, user_id):
    collection = conn.CIA.get_collection("Employees_Info")
    db = await collection.find_one(user_id)
    if db:
        return db['password']

Answer 1

也许你需要安装 pytest-asyncio。这里有更多信息https://fastapi.tiangolo.com/advanced/async-tests/

Answer 2

collection.find_one() 不是异步函数，因此您正在尝试等待作为 dict 的函数的结果，这就是为什么您收到错误 TypeError: object dict can't be used in 'await' expression 您正在等待 dict，而不是协程由异步函数返回。

要修复您的代码，只需从中删除 await

db = await collection.find_one(user_id)

当你这样做时，你并不真的需要它是一个异步函数，所以你可以定期定义它，但你将不得不更改所有函数调用并从中删除await，否则您将再次收到此错误

完整代码：

db.py

def get_user_id(conn, user):
    collection = conn.CIA.get_collection("Employees_Info")
    user = collection.find_one({'name':user},{"_id":1, "name":0, "password":0})
    print(type(user))
    if user:
        return user


def get_password(conn, user_id):
    collection = conn.CIA.get_collection("Employees_Info")
    db = collection.find_one(user_id)
    if db:
        return db['password']

auth.py

def authenticate_user(conn:AsyncIOMotorClient, username, password):
    user_id = verify_user(conn, username)
    if not user_id:
        return False
    
    if not verify_password(conn, user_id, password):
        return False
    
    return user_id


def verify_user(conn, user):
    return get_user_id(conn,user)


def verify_password(conn, user_id, password):
    return pbkdf2_sha256.verify(password, get_password(conn, user_id))

路由器.py

# This probably has to stay as an async function, I'm not sure how the module works
@router.post("/login", tags=["user"], response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), 
                                 db: AsyncIOMotorClient = Depends(get_database)):
    authenticate_user_id = authenticate_user(db, form_data.username, form_data.password)
    if not authenticate_user_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"user_id": str(authenticate_user_id["_id"])})
    return {"access_token": access_token, "token_type": "bearer"}

import collections
from fastapi.testclient import TestClient
from fastapi import status
from main import app
from mongoengine import connect, disconnect, get_connection
from db.db_conn import db


client = TestClient(app)

connect('mongoenginetest', host='mongomock://localhost', alias='testdb')
db.client = get_connection('testdb')
db.client["CIA"]
db.client["Employees_Info"]
db.client.CIA.Employees_Info.insert_one({"name": "user_Name","password": "week"})

def test_ping():
    response = client.get("/")
    assert response.status_code == status.HTTP_200_OK
    assert response.json() == {"message": "Conectado"}
   
def test_login():
    data = {"username":'user_name', 'password':'week'}
    response = client.post("/login", data=data)
    assert response.headers['Content-Type'] == 'application/json'
    assert response.status_code == status.HTTP_200_OK
    db.client.disconnect()