FastAPI，Pytest TypeError:对象dict不能用于'await‘表达式
FastAPI，Pytest TypeError:对象dict不能用于'await‘表达式
提问于 2021-08-31 19:40:32
我正在使用FastAPI编写API当我使用uvicorn运行时一切正常，当我想要使用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文档执行异步测试，但它也不能工作，如果我使用“普通”数据库，测试可以工作。
router.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']回答 2
Stack Overflow用户
发布于 2021-08-31 20:55:05
也许你需要安装pytest-asyncio.Here。这里有更多信息https://fastapi.tiangolo.com/advanced/async-tests/
Stack Overflow用户
发布于 2021-09-01 19:09:29
collection.find_one()不是一个异步函数，因此您正在尝试等待该函数的结果，这就是为什么您得到的错误TypeError: object dict can't be used in 'await' expression您正在等待一个字典，而不是一个异步函数将返回的协程。
要修复您的代码，只需将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))router.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()页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/69004503
复制相关文章
相似问题
腾讯云开发者
