FastAPI属性错误:用户对象没有属性';密码'; [英] FastAPI AttributeError: 'User' object has no attribute 'password'
问题描述
我有一个FastAPI项目,在该项目中我使用了Async SQLAlChemy orm和PostgreSQL作为数据库。基本上,这是一个简单的基于CRUD的工作版块,我已经成功地创建了CRUD操作,他们正在工作,如我所期望的。我遇到的问题是用户身份验证我正在尝试通过JWT在用户注册时实现身份验证,用户将填写字段、用户名、电子邮件和密码,然后将向该用户电子邮件发送一封验证电子邮件,以验证is_active
字段将是True
之后的JWT令牌,默认情况下为False
。我尝试了几种方法,但都不成功,将用户添加到数据库时遇到困难。
routes/route_user.py:
from fastapi import APIRouter, HTTPException, status
from fastapi import Depends
from jose import jwt
from db.models.users import User
from schemas.users import UserCreate, ShowUser
from db.repository.users_data_access_layer import Users
from core.auth import Auth
from core.hashing import Hasher
from core.mailer import Mailer
from core.config import Settings
from depends import get_user_db
router = APIRouter()
get_settings = Settings()
@router.post("/", response_model=ShowUser)
async def create_user(form_data: UserCreate = Depends(), users: Users = Depends(get_user_db)):
if await users.check_user(email=form_data.email) is not None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="User already exists"
)
elif await users.check_username(username=form_data.username) is not None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already exists"
)
new_user = User(email=form_data.email,
username=form_data.username,
hashed_password=Auth.get_password_hash(form_data.password)
)
await users.register_user(new_user)
print(new_user)
confirmation = Auth.get_confirmation_token(new_user.id)
print(confirmation)
new_user.confirmation = confirmation["jti"]
try:
Mailer.send_confirmation_message(confirmation["token"], form_data.email)
except ConnectionRefusedError:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Email couldn't be send. Please try again."
)
return await users.register_user(form_data)
@router.get("/verify/{token}")
async def verify(token: str, users: Users = Depends(get_user_db)):
invalid_token_error = HTTPException(status_code=400, detail="Invalid Token")
try:
payload = jwt.decode(token, get_settings.SECRET_KEY, algorithms=[get_settings.TOKEN_ALGORITHM])
print(payload['sub'])
except jwt.JWSError:
raise HTTPException(status_code=403, detail="Token has Expired")
if payload['scope'] != 'registration':
raise invalid_token_error
print(payload['sub'])
user = await users.get_user_by_id(id=payload['sub'])
print(user)
print('hello2')
if not user or await users.get_confirmation_uuid(str(User.confirmation)) != payload['jti']:
print('hello')
raise invalid_token_error
if user.is_active:
print('hello2')
raise HTTPException(status_code=403, detail="User already Activated")
user.confirmation = None
user.is_active = True
return await users.register_user(user)
上述路由输出属性错误:
File ".db
epositoryusers_data_access_layer.py", line 26, in register_user
hashed_password=user.password,
AttributeError: 'User' object has no attribute 'password'
user_data_access_layer.py
这是所有数据库通信发生的地方。为了方便起见,我想我需要一些save method
添加到db中,但我不知道如何实现它。我尝试了这样的操作:
from core.hashing import Hasher
from sqlalchemy.orm import Session
from sqlalchemy.sql.expression import select
from sqlalchemy.sql import exists
from db.models.users import User
from schemas.users import UserCreate
from core.hashing import Hasher
db_session = Session
class Users():
def __init__(self, db_session: Session):
self.db_session = db_session
async def save(self):
if self.id == None:
self.db_session.add(self)
return await self.db_session.flush()
#print('user created')
async def register_user(self, user: UserCreate):
new_user = User(username=user.username,
email=user.email,
hashed_password=user.password,
is_active = False,
is_superuser=False
)
self.db_session.add(new_user)
await self.db_session.flush()
return new_user
async def check_user(self, email: str):
user_exist = await self.db_session.execute(select(User).filter(User.email==email))
#print(user_exist)
return user_exist.scalar_one_or_none()
async def check_username(self, username: str):
user_exist = await self.db_session.execute(select(User).filter(User.username==username))
#print(user_exist)
return user_exist.scalar_one_or_none()
async def get_user_by_id(self, id: int):
user_exist = await self.db_session.execute(select(User).filter(User.id==id)
#print(user_exist)
return user_exist.scalar_one_or_none()
async def get_confirmation_uuid(self, confirmation_uuid:str):
user_exist = await self.db_session.execute(select(User).filter(str(User.confirmation)==confirmation_uuid))
#print(user_exist)
return user_exist
架构/users.py
from typing import Optional
from pydantic import BaseModel, EmailStr
class UserBase(BaseModel):
username: str
email: EmailStr
password: str
class UserCreate(UserBase):
username: str
email: EmailStr
password: str
class ShowUser(UserBase):
username: str
email: EmailStr
is_active: bool
class Config():
orm_mode = True
Models/users.py
import uuid
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy import Column, Integer, String, Boolean, ForeignKey
from sqlalchemy.orm import relationship
from db.base_class import Base
class User(Base):
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
username = Column(String, unique=True, nullable=False)
email = Column(String, nullable=False, unique=True, index=True)
hashed_password = Column(String(255), nullable=False)
is_active = Column(Boolean, default=False)
is_superuser = Column(Boolean, default=False)
confirmation = Column(UUID(as_uuid=True), nullable=True, default=uuid.uuid4)
jobs = relationship("Job", back_populates="owner")
依赖项
from db.session import async_session
from db.repository.jobs_data_access_layer import JobBoard
from db.repository.users_data_access_layer import Users
async def get_job_db():
async with async_session() as session:
async with session.begin():
yield JobBoard(session)
async def get_user_db():
async with async_session() as session:
async with session.begin():
yield Users(session)
因为这都是新事物,无论我到达哪里,我都遇到了困难,我已经在这个项目上工作了几个星期,还找不到解决办法,所以如果有任何帮助,我将不胜感激。
推荐答案
代码中存在不同的问题。首先,使用错误的参数调用一些对类模型/用户的方法的调用。实际上,有些是使用User对象作为参数来调用的,而那些需要的则是一个Pydtic UserCreate模型。因此,当您发送User
对象而不是PYDANIC模型时,password
属性不存在。
其次,之后还会出现其他问题,因为检索User对象的方法实际上返回一个列表(ChunkIterator)。但是,您可以像接收对象一样进行比较。
我擅自提出了一个替代方案,重新设计了您的一些代码。
现在,我已经创建了一些方法来保存数据库中用户的最终修改,并创建了一些方法,这些方法根据不同的条件(id、用户名、电子邮件)返回用户,但与您的代码相反,这些方法返回对象(或无)而不是列表。
User_Data_Access_layer.py
from fastapi import HTTPException, status
from db.models.users import User
from schemas.users import UserCreate
from db_config import SESSION
from auth import Auth
class Users():
def __init__(self):
pass
@classmethod
async def save(cls, user_instance):
try:
SESSION.add(user_instance)
SESSION.commit()
except Exception as error:
SESSION.rollback()
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
@classmethod
async def get_user_by_id(cls, id):
user = await SESSION.query(User).filter(User.id == id).one_or_none()
return user
@classmethod
async def get_user_by_username(cls, username):
user = await SESSION.query(User).filter(User.username == username).one_or_none()
return user
@classmethod
async def get_user_by_email(cls, email):
user = await SESSION.query(User).filter(User.email == email).one_or_none()
return user
@classmethod
async def get_user_by_confirmation(cls, confirmation):
user = await SESSION.query(User).filter(User.confirmation == confirmation).one_or_none()
return user
@classmethod
async def create_user(self, user: UserCreate):
new_user = User(username=user.username,
email=user.email,
hashed_password=Auth.get_password_hash(user.password)
)
cls.save(new_user)
return new_user
如您所见,我从您的层文件中删除了会话创建,并将其放入一个全局变量中,放在一个单独的文件中。
db_config.py
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.session import Session
ENGINE: Engine = create_engine(your_config_url, pool_pre_ping=True)
SESSION: Session = sessionmaker(bind=ENGINE)()
最后,这里是根据建议的代码更新的终结点。我在其中添加了注释以使其更易于理解。
route.py
@router.post("/", response_model=ShowUser)
async def create_user(form_data: UserCreate = Depends(), users: Users = Depends(get_user_db)):
# CHECK IF USER ALREADY EXISTS
if await Users.get_user_by_email(email=form_data.email) is not None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="User already exists"
)
# CHECK IF USERNAME ALREADY EXISTS
elif await Users.get_user_by_username(username=form_data.username) is not None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already exists"
)
# CREATE USER WITH USERS METHOD
# now the hashing of the password is done directly in the creation method
new_user = await Users.create_user(form_data)
# GET TOKEN
# we no longer create a new uid for JTI, but use the one created automatically during user creation
# so I modified the get_confirmation_token function so that it takes the user's JTI uid as a parameter
confirmation_token = Auth.get_confirmation_token(
new_user.id,
new_user.confirmation)
这篇关于FastAPI属性错误:用户对象没有属性';密码';的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!