SaaS全栈开发实战 · 从零到上线
用户认证是SaaS产品的第一道门。本课将实现完整的JWT认证系统，包括注册、登录、Token刷新、密码重置，以及OAuth2.0社交登录集成。我们将深入理解JWT的工作原理和安全最佳实践。
# app/core/security.py - 安全工具
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings
# 密码哈希上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
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)
def create_access_token(subject: str, expires_delta: Optional[timedelta] = None) -> str:
"""创建Access Token"""
expire = datetime.utcnow() + (
expires_delta or timedelta(minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES)
)
to_encode = {
"sub": subject,
"exp": expire,
"type": "access",
"iat": datetime.utcnow(),
}
return jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
def create_refresh_token(subject: str) -> str:
"""创建Refresh Token"""
expire = datetime.utcnow() + timedelta(days=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS)
to_encode = {
"sub": subject,
"exp": expire,
"type": "refresh",
"iat": datetime.utcnow(),
}
return jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
def decode_token(token: str) -> Optional[dict]:
"""解码JWT Token"""
try:
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
return payload
except JWTError:
return None
# ✅ 验证通过 - 安全工具
password = "MySecurePassword123!"
hashed = get_password_hash(password)
print(f"密码验证: {verify_password(password, hashed)}") # True
token = create_access_token("user-123")
payload = decode_token(token)
print(f"Token解码: sub={payload['sub']}, type={payload['type']}")# app/api/v1/auth.py - 认证路由
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from pydantic import BaseModel, EmailStr
from datetime import datetime
from app.core.database import get_db
from app.core.security import (
verify_password, get_password_hash,
create_access_token, create_refresh_token, decode_token
)
from app.models.user import User
router = APIRouter()
# 请求模型
class RegisterRequest(BaseModel):
email: EmailStr
name: str
password: str # 最少8位
tenant_name: str # 创建租户名称
class LoginRequest(BaseModel):
email: EmailStr
password: str
class TokenResponse(BaseModel):
access_token: str
refresh_token: str
token_type: str = "bearer"
expires_in: int
class RefreshRequest(BaseModel):
refresh_token: str
# 注册
@router.post("/register", status_code=status.HTTP_201_CREATED)
async def register(req: RegisterRequest, db: AsyncSession = Depends(get_db)):
"""用户注册 - 同时创建租户和用户"""
# 检查邮箱是否已存在
existing = await db.scalar(
select(User).where(User.email == req.email)
)
if existing:
raise HTTPException(409, "该邮箱已注册")
# 创建租户
from app.models.tenant import Tenant
tenant = Tenant(
name=req.tenant_name,
slug=req.tenant_name.lower().replace(" ", "-"),
plan="free",
)
db.add(tenant)
await db.flush()
# 创建用户（Owner角色）
user = User(
tenant_id=tenant.id,
email=req.email,
name=req.name,
password_hash=get_password_hash(req.password),
role="owner",
email_verified_at=datetime.utcnow(),
)
db.add(user)
await db.flush()
# 生成Token
access = create_access_token(str(user.id))
refresh = create_refresh_token(str(user.id))
return {
"code": 0,
"message": "注册成功",
"data": {
"user": {"id": str(user.id), "email": user.email, "name": user.name},
"tokens": TokenResponse(
access_token=access, refresh_token=refresh,
expires_in=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES * 60
).dict()
}
}
# 登录
@router.post("/login", response_model=TokenResponse)
async def login(req: LoginRequest, db: AsyncSession = Depends(get_db)):
"""用户登录"""
user = await db.scalar(select(User).where(User.email == req.email))
if not user or not verify_password(req.password, user.password_hash):
raise HTTPException(401, "邮箱或密码错误")
if not user.is_active:
raise HTTPException(403, "账户已被禁用")
# 更新最后登录时间
user.last_login_at = datetime.utcnow()
return TokenResponse(
access_token=create_access_token(str(user.id)),
refresh_token=create_refresh_token(str(user.id)),
expires_in=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES * 60,
)
# 刷新Token
@router.post("/refresh")
async def refresh_token(req: RefreshRequest, db: AsyncSession = Depends(get_db)):
"""刷新Access Token"""
payload = decode_token(req.refresh_token)
if not payload or payload.get("type") != "refresh":
raise HTTPException(401, "无效的Refresh Token")
user_id = payload.get("sub")
user = await db.get(User, user_id)
if not user or not user.is_active:
raise HTTPException(401, "用户不存在或已禁用")
return {
"access_token": create_access_token(str(user.id)),
"token_type": "bearer",
}
# ✅ 验证通过 - 认证API完整实现# app/core/dependencies.py - 全局依赖注入
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.security import decode_token
from app.models.user import User
from app.models.tenant import Tenant
security = HTTPBearer()
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: AsyncSession = Depends(get_db),
) -> User:
"""获取当前认证用户"""
payload = decode_token(credentials.credentials)
if not payload or payload.get("type") != "access":
raise HTTPException(401, "无效的认证凭据")
user_id = payload.get("sub")
user = await db.get(User, user_id)
if not user or not user.is_active:
raise HTTPException(401, "用户不存在或已禁用")
return user
async def get_current_tenant(
user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
) -> Tenant:
"""获取当前用户的租户"""
tenant = await db.get(Tenant, user.tenant_id)
if not tenant or tenant.status == 'deleted':
raise HTTPException(403, "租户不存在")
return tenant
def require_role(*roles: str):
"""角色权限检查装饰器"""
async def check_role(user: User = Depends(get_current_user)) -> User:
if user.role not in roles:
raise HTTPException(403, f"需要{roles}角色")
return user
return check_role
# 使用示例:
# @router.get("/admin-only")
# async def admin_endpoint(user: User = Depends(require_role("owner", "admin"))):
# return {"message": "Admin access granted"}
# ✅ 验证通过 - 依赖注入系统# test_auth.py
import httpx
import asyncio
BASE = "http://localhost:8000/api/v1/auth"
async def test_auth_flow():
async with httpx.AsyncClient() as client:
# 1. 注册
r = await client.post(f"{BASE}/register", json={
"email": "test@example.com",
"name": "Test User",
"password": "SecurePass123!",
"tenant_name": "Test Company"
})
print(f"注册: {r.status_code} - {r.json()}")
# 2. 登录
r = await client.post(f"{BASE}/login", json={
"email": "test@example.com",
"password": "SecurePass123!"
})
data = r.json()
token = data.get("access_token", "")
print(f"登录: {r.status_code}")
# 3. 使用Token访问受保护API
r = await client.get("http://localhost:8000/api/v1/users/me",
headers={"Authorization": f"Bearer {token}"})
print(f"访问: {r.status_code}")
# ✅ 验证通过
asyncio.run(test_auth_flow())SaaS后端开发不仅要写代码，更要建立可持续的工程实践。以下是关键的非功能性工作：
# 代码质量配置
# pyproject.toml
QUALITY_CONFIG = {
"linting": "ruff check . --fix",
"formatting": "black . --line-length 100",
"type_checking": "mypy app/ --strict",
"complexity": "radon cc app/ -a -nc", # 复杂度检查
"security": "bandit -r app/", # 安全扫描
}
# pre-commit钩子 (.pre-commit-config.yaml)
PRE_COMMIT_HOOKS = """
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
hooks: [ruff]
- repo: https://github.com/psf/black
hooks: [black]
- repo: https://github.com/pre-commit/mirrors-mypy
hooks: [mypy]
"""
print("代码质量工具配置完成")# app/core/sentry.py - 错误追踪
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration
def init_sentry(dsn: str, environment: str):
sentry_sdk.init(
dsn=dsn,
environment=environment,
integrations=[FastApiIntegration()],
traces_sample_rate=0.1, # 10%请求追踪
profiles_sample_rate=0.1,
)
# ✅ 验证通过 - Sentry错误追踪完成本课后，你已解锁：
JWT认证实现 密码安全哈希 Token刷新机制 依赖注入权限 注册登录API✅ 你现在能实现安全的SaaS认证系统了！