FastAPI属性错误：用户对象没有属性&#39；密码&#39； [英] 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属性错误：用户对象没有属性&#39；密码&#39；的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！