综合项目实战：电子商务API用户认证与权限系统

项目简介

在电子商务应用中，用户认证和权限系统是核心组成部分，用于保护用户数据和业务逻辑。本教程将带领你使用FastAPI构建一个完整的电子商务API，专注于实现用户认证（如登录、注册）和权限管理（如基于角色的访问控制）。通过这个实战项目，你将掌握FastAPI的高级特性，并应用到实际场景中。

环境准备

在开始之前，确保你的开发环境已经设置好：

• Python 3.7或更高版本：FastAPI支持现代Python版本。

• 安装FastAPI和相关依赖：使用pip安装所需包。

  pip install fastapi uvicorn sqlalchemy pydantic[email] python-jose[cryptography] passlib[bcrypt]
  

  • fastapi 和 uvicorn：用于创建和运行API服务器。
  • sqlalchemy：用于ORM（对象关系映射）来操作数据库。
  • pydantic：用于数据验证和设置。
  • python-jose 和 passlib：用于JWT认证和密码哈希。

• 数据库：我们将使用SQLite作为示例，因为它易于设置。在生产环境中，你可能需要切换到PostgreSQL或MySQL。

项目结构

创建一个项目文件夹，并组织文件结构如下：

ecommerce_auth_project/
├── main.py           # 主要FastAPI应用文件
├── models.py         # 数据库模型定义
├── schemas.py        # Pydantic模型用于数据验证
├── auth.py           # 认证相关函数（JWT、密码哈希等）
├── dependencies.py   # 依赖注入（如获取当前用户）
├── crud.py           # 数据库操作函数
├── database.py       # 数据库连接设置
└── routers/          # 路由模块
    ├── users.py      # 用户相关路由
    └── items.py      # 商品相关路由（示例，用于权限测试）

步骤1：设置数据库模型

在models.py中，定义用户和角色模型。

from sqlalchemy import Column, Integer, String, Boolean, ForeignKey
from sqlalchemy.orm import relationship
from database import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)
    role_id = Column(Integer, ForeignKey("roles.id"))
    role = relationship("Role", back_populates="users")

class Role(Base):
    __tablename__ = "roles"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, unique=True, index=True)  # 如 "admin", "user"
    users = relationship("User", back_populates="role")

在database.py中，设置数据库引擎和会话。

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./ecommerce.db"  # 使用SQLite数据库
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

步骤2：实现用户认证

在auth.py中，添加JWT认证和密码哈希功能。

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import HTTPException, status

SECRET_KEY = "your-secret-key"  # 在生产环境中，使用环境变量安全存储
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email: str = payload.get("sub")
        if email is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
        return email
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

步骤3：创建Pydantic模型

在schemas.py中，定义数据验证模型。

from pydantic import BaseModel, EmailStr
from typing import Optional

class UserCreate(BaseModel):
    email: EmailStr
    password: str
    role_id: Optional[int] = 2  # 默认角色为普通用户，假设role_id 2是"user"

class UserLogin(BaseModel):
    email: EmailStr
    password: str

class Token(BaseModel):
    access_token: str
    token_type: str

步骤4：实现CRUD操作

在crud.py中，添加数据库操作函数。

from sqlalchemy.orm import Session
from models import User
from auth import get_password_hash, verify_password

def create_user(db: Session, user: UserCreate):
    hashed_password = get_password_hash(user.password)
    db_user = User(email=user.email, hashed_password=hashed_password, role_id=user.role_id)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_user_by_email(db: Session, email: str):
    return db.query(User).filter(User.email == email).first()

def authenticate_user(db: Session, email: str, password: str):
    user = get_user_by_email(db, email)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

步骤5：添加依赖注入

在dependencies.py中，创建获取当前用户的依赖。

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from database import get_db
from auth import verify_token
from models import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    email = verify_token(token)
    user = db.query(User).filter(User.email == email).first()
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
    return user

步骤6：实现权限系统

在routers/users.py中，添加用户相关路由，包括注册、登录和获取用户信息。

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from database import get_db
from schemas import UserCreate, UserLogin, Token
from crud import create_user, authenticate_user
from auth import create_access_token

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/register", response_model=Token)
async def register(user: UserCreate, db: Session = Depends(get_db)):
    # 检查邮箱是否已存在
    db_user = db.query(User).filter(User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    new_user = create_user(db, user)
    access_token = create_access_token(data={"sub": new_user.email})
    return {"access_token": access_token, "token_type": "bearer"}

@router.post("/login", response_model=Token)
async def login(user_login: UserLogin, db: Session = Depends(get_db)):
    user = authenticate_user(db, user_login.email, user_login.password)
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect email or password")
    access_token = create_access_token(data={"sub": user.email})
    return {"access_token": access_token, "token_type": "bearer"}

为了演示权限，创建一个routers/items.py，仅允许特定角色访问。

from fastapi import APIRouter, Depends, HTTPException, status
from dependencies import get_current_user
from models import User, Role

router = APIRouter(prefix="/items", tags=["items"])

def check_admin_permission(current_user: User = Depends(get_current_user)):
    if current_user.role.name != "admin":
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
    return current_user

@router.get("/")
async def read_items(current_user: User = Depends(check_admin_permission)):
    return {"message": "Only admin can view items"}

步骤7：整合并运行应用

在main.py中，集成所有模块并启动服务器。

from fastapi import FastAPI
from database import engine, Base
from routers import users, items

# 创建数据库表
Base.metadata.create_all(bind=engine)

app = FastAPI(title="ECommerce Auth API", description="A user authentication and permission system for e-commerce")

# 包含路由
app.include_router(users.router)
app.include_router(items.router)

@app.get("/")
async def root():
    return {"message": "Welcome to the ECommerce Auth API"}

运行应用：

uvicorn main:app --reload

访问 http://127.0.0.1:8000/docs 查看交互式API文档。

步骤8：测试和部署

• 测试：使用FastAPI的TestClient编写单元测试，确保认证和权限工作正常。
• 安全考虑：
  • 使用环境变量存储SECRET_KEY。
  • 在生产环境中启用HTTPS。
  • 定期更新依赖包。
• 部署建议：
  • 使用Docker容器化应用。
  • 部署到云平台如AWS、Google Cloud或Azure。
  • 设置监控和日志记录。

总结

通过本教程，你学会了如何使用FastAPI构建一个电子商务API的用户认证与权限系统。关键点包括：

• 设置FastAPI和数据库模型。
• 实现JWT认证和密码哈希。
• 使用Pydantic进行数据验证。
• 创建依赖注入来处理当前用户和权限检查。
• 整合路由并测试API。

这个项目可以作为基础，扩展更多功能如商品管理、订单处理等。继续探索FastAPI文档以深入学习。

祝你学习愉快！如有问题，欢迎查阅FastAPI官方文档或社区资源。