FastAPI AttributeError: 'User' object has no attribute 'password'
FastAPI AttributeError: 'User' object has no attribute 'password'
我有一个 FastAPI 项目，我在其中使用 Async SQLAlchemy orm 和 postgresql 作为数据库。基本上它是一个简单的基于 CRUD 的工作板，我已经成功地创建了 CRUD 操作，并且它们按我预期的那样工作。我偶然发现的问题是用户身份验证我正在尝试通过 JWT 在用户注册时实施身份验证，用户将填写字段、用户名、电子邮件和密码，然后验证电子邮件将发送到该用户电子邮件以验证 JWT is_active 字段之后的标记将是 True，默认情况下是 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\repository\users_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 来添加到数据库中以方便使用，但我不知道如何实现它。我试过这样的事情：
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
schemas/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")
depends.py
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)
因为这是全新的东西，无论我走到哪里，我都碰壁了，我已经在这个项目上工作了几个星期了，但还找不到解决它的方法，所以任何帮助都将不胜感激。
table没有密码，table只有hashed_password
new_user = User(username=user.username,
email=user.email,
hashed_password=user.hashed_password,
is_active = False,
is_superuser=False
)
代码中存在不同的问题。首先，一些对 class model/Users 方法的调用是用错误的参数调用的。实际上，有些调用时使用 User 对象作为参数，而那些期望使用 Pydantic UserCreate 模型。因此，当您发送 User 对象而不是 Pydantic 模型时，password 属性不存在。
其次，之后会出现其他问题，因为您检索用户对象的方法实际上 return 一个列表（ChunkIterator）。但是，您进行比较时就好像收到一个对象一样。
我冒昧地通过重新编写您的一些代码来提出替代方案。
现在，我已经创建了一些方法来在您的数据库中保存用户的最终修改，并创建了一些方法 return 用户根据不同的标准（id、用户名、电子邮件）除了与您的代码相反，那些 return 对象（或 None）而不是列表。
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
如您所见，我从您的层文件中删除了 Session 创建，并将其放在一个全局变量中，在一个单独的文件中。
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 项目，我在其中使用 Async SQLAlchemy orm 和 postgresql 作为数据库。基本上它是一个简单的基于 CRUD 的工作板，我已经成功地创建了 CRUD 操作，并且它们按我预期的那样工作。我偶然发现的问题是用户身份验证我正在尝试通过 JWT 在用户注册时实施身份验证，用户将填写字段、用户名、电子邮件和密码，然后验证电子邮件将发送到该用户电子邮件以验证 JWT is_active 字段之后的标记将是 True，默认情况下是 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\repository\users_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 来添加到数据库中以方便使用，但我不知道如何实现它。我试过这样的事情：
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
schemas/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")
depends.py
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)
因为这是全新的东西，无论我走到哪里，我都碰壁了，我已经在这个项目上工作了几个星期了，但还找不到解决它的方法，所以任何帮助都将不胜感激。
table没有密码，table只有hashed_password
new_user = User(username=user.username,
email=user.email,
hashed_password=user.hashed_password,
is_active = False,
is_superuser=False
)
代码中存在不同的问题。首先，一些对 class model/Users 方法的调用是用错误的参数调用的。实际上，有些调用时使用 User 对象作为参数，而那些期望使用 Pydantic UserCreate 模型。因此，当您发送 User 对象而不是 Pydantic 模型时，password 属性不存在。
其次，之后会出现其他问题，因为您检索用户对象的方法实际上 return 一个列表（ChunkIterator）。但是，您进行比较时就好像收到一个对象一样。
我冒昧地通过重新编写您的一些代码来提出替代方案。
现在，我已经创建了一些方法来在您的数据库中保存用户的最终修改，并创建了一些方法 return 用户根据不同的标准（id、用户名、电子邮件）除了与您的代码相反，那些 return 对象（或 None）而不是列表。
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
如您所见，我从您的层文件中删除了 Session 创建，并将其放在一个全局变量中，在一个单独的文件中。
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)