CODEWITHSUNDEEP
pythonpython-3.xauthenticationjwtfastapi
dougj
dougj

Reputation: 135

FastAPI User Authentication with verification Email

I am creating the User Authentication API route that is able to register the user if it is not already registered in the database and if an email address is in the database it will alert the user with User already exists.

I'm trying to send the user verification email to verify the JWT token on registration, But for some reasons I couldn't able to save the user in the database whenever I try to make a new user it keeps giving me User already exists.

Here is the API structure:

user_data_access_layer.py

from sqlalchemy.orm import Session
from sqlalchemy.sql.expression import select
from sqlalchemy.sql import exists


from schemas.users import UserCreate
from db.models.users import User


db_session = Session
class Users():
    
    def __init__(self, db_session: Session):
        self.db_session = db_session

    async def save_user(self, user: UserCreate):
        if self.id == None:
            self.db_session.add(self)
            return await self.db_session.flush()

    async def check_user(self, email: str):
        user_exist = await self.db_session.execute(select(exists(User.email==email)))
        user = user_exist.scalar_one_or_none()
        return user

route_user.py

from fastapi import APIRouter, HTTPException, status
from fastapi import Depends

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"
        )

    user = User(email=form_data.email,
                hashed_password=Hasher.get_password_hash(form_data.password),
                is_active = False,
                is_superuser=False)

    confirmation = Auth.get_confirmation_token(user.id)
    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.save_user(user)

models.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(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, nullable=False)
    email = Column(String, nullable=False, unique=True, index=True)
    hashed_password = Column(String, 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")

schemas/users.py

from typing import Optional
from pydantic import BaseModel, EmailStr


class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str

class ShowUser(BaseModel):
    username: str
    email: EmailStr
    is_active: bool

    class Config():
        orm_mode = True

Upvotes: 0

Views: 7331

Answers (1)

manuel220
manuel220

Reputation: 1

if you are using SQLAlchemy 1.4+ try this instead:

user_exist = await self.db_session.scalar(select(User).where(User.email==email))

Upvotes: 0

Related Questions