FastAPI AttributeError: 'User' object has no attribute 'password'
I have a FastAPI project where I use Async SQLAlchemy orm and postgresql as database. Basically it's a simple CRUD based job board, I have successfully created CRUD operations and they work as I expect. The problem I stumbled upon is user authentication I am trying to implement authentication via JWT when the user registers, the user will fill in the fields, username, email and password, then a verification email will be sent to that user email to validate the JWT Tokens after this is_active
field will be True
by default False
. I've tried several ways without success, I'm having a hard time adding users to the database.
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)
The above route output attribute is wrong:
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
This is where all database communication happens. Here, I think I need to add something to the db save method
for convenience , but I don't know how to implement it. I have tried something like this:
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
schema/user.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")
dependencies.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)
Since this is brand new, wherever I go, I run into a wall, I've been working on this project for a few weeks now and haven't found a way around it, so any help would be greatly appreciated.
There are different problems in the code. First, some calls to the methods of the class model/user were called with the wrong parameters. In fact, some are called with User objects as parameters, while others require a Pydantic UserCreate model. So when you send an User
object instead of a Pydantic model, the password
property doesn't exist. Second, other problems arise later because your method of retrieving the User object actually returns a list (ChunkIterator). However, you compare as if you were receiving an object.
I took the liberty of suggesting an alternative by rewriting some of your code.
Now, I've created some methods to save the user's final modification to the database, and some methods that return the user based on different criteria (id, username, email), except that contrary to your code, those that return an object (or none ) instead of a list.
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
As you can see, I removed the Session creation from your layer file and put it in a global variable in a separate file.
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)()
All in all, this is the updated endpoint based on the suggested code. I've added comments to make it easier to understand.
routes.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)