python - 带有验证电子邮件的 FastAPI 用户身份验证

问题描述

我正在创建用户身份验证 API 路由，如果它尚未在数据库中注册，并且如果电子邮件地址在数据库中，它将能够注册用户，它会用User already exists.

我正在尝试发送用户验证电子邮件以验证注册时的 JWT 令牌，但由于某些原因，每当我尝试创建一个新用户时，我都无法将用户保存在数据库中User already exists。

下面是 API 结构：

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)

模型.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")

架构/用户.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

标签： pythonpython-3.xauthenticationjwtfastapi

python - 带有验证电子邮件的 FastAPI 用户身份验证

问题描述

解决方案

推荐阅读