【发布时间】：2021-09-23 21:09:41
【问题描述】：
我有一个 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 以方便添加到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")
取决于.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)
因为这是全新的东西，无论我走到哪里，我都碰壁了，我已经在这个项目上工作了几个星期，但还没有找到解决方法，所以任何帮助都将不胜感激。
【问题讨论】：
标签： python postgresql sqlalchemy fastapi