用户认证的魔法配方：从模型设计到密码安全的奇幻之旅
title: 用户认证的魔法配方：从模型设计到密码安全的奇幻之旅
from datetime import datetime
from uuid import uuid4
from sqlalchemy import Column, String, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(String(36), primary_key=True, default=lambda: str(uuid4()))
username = Column(String(50), unique=True, nullable=False)
email = Column(String(255), unique=True, nullable=False)
hashed_password = Column(String(255), nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
def __repr__(self):
return f"<User {self.username}>"
# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False
)
async def get_db():
async with AsyncSessionLocal() as session:
yield session
# 初始化迁移环境
alembic init migrations
# 生成迁移文件
alembic revision --autogenerate -m "create users table"
# 执行迁移
alembic upgrade head
# security.py
from passlib.context import CryptContext
pwd_context = CryptContext(
schemes=["bcrypt"],
deprecated="auto",
bcrypt__rounds=12
)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
class User(Base):
# ...其他字段同上
def set_password(self, password: str):
self.hashed_password = get_password_hash(password)
def check_password(self, password: str) -> bool:
return verify_password(password, self.hashed_password)
from pydantic import BaseModel, constr
class UserCreate(BaseModel):
username: constr(min_length=4, max_length=50)
email: str
password: constr(min_length=8)
class UserLogin(BaseModel):
username: str
password: str
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
router = APIRouter()
@router.post("/register")
async def register(
user_data: UserCreate,
db: AsyncSession = Depends(get_db)
):
# 检查用户名是否已存在
existing_user = await db.execute(
select(User).where(User.username == user_data.username)
)
if existing_user.scalars().first():
raise HTTPException(400, "Username already registered")
# 创建用户对象
new_user = User(
username=user_data.username,
email=user_data.email
)
new_user.set_password(user_data.password)
# 保存到数据库
db.add(new_user)
await db.commit()
await db.refresh(new_user)
return {"message": "User created successfully"}
fastapi==0.68.2
uvicorn==0.15.0
sqlalchemy==1.4.35
asyncpg==0.24.0
passlib==1.7.4
python-multipart==0.0.5
alembic==1.7.5
pydantic==1.8.2
pip install fastapi uvicorn sqlalchemy asyncpg passlib python-multipart alembic pydantic
评论
发表评论