第04课：数据库设计

SaaS全栈开发实战 · 从零到上线

📖 课程概述

数据库设计是SaaS架构的核心。糟糕的数据库设计让后续开发步履维艰，优秀的设计让扩展自然而然。本课深入SaaS特有的数据库挑战：多租户数据隔离、订阅计费模型、审计追踪，以及PostgreSQL高级特性。

🏗️ SaaS数据库设计原则

• 租户隔离优先：数据隔离是SaaS的底线安全要求
• 软删除为王：SaaS数据永远不要真删除
• 审计追踪完整：每条记录的创建和修改必须可追溯
• JSONB辅助扩展：灵活配置避免频繁加列
• 金额用分存储：避免浮点精度问题

📐 核心ER图

💻 完整数据库Schema

复制-- ============================================
-- SaaS Platform Database Schema (PostgreSQL 15+)
-- ============================================
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

-- 租户管理
CREATE TYPE tenant_status AS ENUM ('trial', 'active', 'suspended', 'deleted');
CREATE TABLE tenants (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    name VARCHAR(255) NOT NULL,
    slug VARCHAR(100) UNIQUE NOT NULL,
    status tenant_status DEFAULT 'trial',
    plan VARCHAR(50) DEFAULT 'free',
    settings JSONB DEFAULT '{"locale":"zh-CN","timezone":"Asia/Shanghai"}'::jsonb,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    deleted_at TIMESTAMPTZ  -- 软删除
);

-- 自动更新时间戳
CREATE OR REPLACE FUNCTION update_timestamp()
RETURNS TRIGGER AS $$
BEGIN NEW.updated_at = NOW(); RETURN NEW; END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER tenants_updated_at
    BEFORE UPDATE ON tenants
    FOR EACH ROW EXECUTE FUNCTION update_timestamp();

-- 用户管理
CREATE TYPE user_role AS ENUM ('owner', 'admin', 'member', 'viewer');
CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    email VARCHAR(255) NOT NULL,
    name VARCHAR(255) NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    role user_role DEFAULT 'member',
    is_active BOOLEAN DEFAULT TRUE,
    email_verified_at TIMESTAMPTZ,
    last_login_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    deleted_at TIMESTAMPTZ
);

-- 租户内邮箱唯一
CREATE UNIQUE INDEX idx_users_email_tenant 
    ON users(email, tenant_id) WHERE deleted_at IS NULL;
CREATE INDEX idx_users_tenant ON users(tenant_id) WHERE deleted_at IS NULL;

-- 订阅与计费
CREATE TYPE subscription_status AS ENUM (
    'trial', 'active', 'past_due', 'canceled', 'expired'
);
CREATE TABLE subscriptions (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    plan VARCHAR(50) NOT NULL,
    status subscription_status DEFAULT 'trial',
    stripe_customer_id VARCHAR(255),
    stripe_subscription_id VARCHAR(255),
    current_period_start TIMESTAMPTZ,
    current_period_end TIMESTAMPTZ,
    canceled_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_subscriptions_tenant ON subscriptions(tenant_id);

-- 账单
CREATE TYPE invoice_status AS ENUM ('draft','open','paid','void','uncollectible');
CREATE TABLE invoices (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    subscription_id UUID REFERENCES subscriptions(id),
    amount_cents INTEGER NOT NULL,  -- 用分存储！
    currency VARCHAR(3) DEFAULT 'USD',
    status invoice_status DEFAULT 'open',
    stripe_invoice_id VARCHAR(255),
    due_date DATE NOT NULL,
    paid_at TIMESTAMPTZ,
    line_items JSONB DEFAULT '[]'::jsonb,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- ✅ 验证通过 - PostgreSQL DDL

🔐 多租户数据隔离策略

复制# 三种多租户隔离策略
class TenantIsolation:
    STRATEGIES = {
        'schema_per_tenant': {
            'name': 'Schema隔离',
            'isolation': '强', 'cost': '中', 'max_tenants': 1000,
            'pros': ['物理隔离', '可单独备份'],
            'cons': ['迁移复杂', '连接池管理难'],
            'best_for': '大型B2B SaaS',
        },
        'row_level_security': {
            'name': '行级安全(RLS)',
            'isolation': '中', 'cost': '低', 'max_tenants': 100000,
            'pros': ['运维简单', '成本最低'],
            'cons': ['性能开销', '调试困难'],
            'best_for': '中小型SaaS快速起步',
        },
        'hybrid': {
            'name': '混合策略',
            'isolation': '弹性', 'cost': '中', 'max_tenants': '无限',
            'pros': ['最优性价比', '弹性伸缩'],
            'cons': ['架构复杂', '两套代码路径'],
            'best_for': '有Enterprise客户的SaaS',
        }
    }

# ✅ 验证通过
for k, v in TenantIsolation.STRATEGIES.items():
    print(f"📋 {v['name']}: 隔离={v['isolation']}, 最大租户={v['max_tenants']}")

🛡️ RLS实战

复制-- PostgreSQL RLS - 行级安全策略

-- 1. 启用RLS
ALTER TABLE users ENABLE ROW LEVEL SECURITY;
ALTER TABLE subscriptions ENABLE ROW LEVEL SECURITY;
ALTER TABLE invoices ENABLE ROW LEVEL SECURITY;

-- 2. 创建隔离策略
CREATE POLICY tenant_isolation_users ON users
    USING (tenant_id = current_setting('app.current_tenant')::UUID);

CREATE POLICY tenant_isolation_subscriptions ON subscriptions
    USING (tenant_id = current_setting('app.current_tenant')::UUID);

-- 3. 超级管理员可跨租户查询
CREATE POLICY superuser_bypass ON users
    USING (current_setting('app.is_superuser', true) = 'true');

-- 4. 应用层中间件设置租户上下文
-- FastAPI中间件:
-- SET app.current_tenant = 'tenant-uuid-here';

-- ✅ 验证通过 - RLS策略

🚀 部署步骤

复制cd saas-project/backend
alembic init alembic
alembic revision --autogenerate -m "initial_schema"
alembic upgrade head

复制# app/models/tenant.py
from sqlalchemy import Column, String, Enum, DateTime, Boolean
from sqlalchemy.dialects.postgresql import UUID, JSONB
from app.core.database import Base
import uuid

class Tenant(Base):
    __tablename__ = "tenants"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = Column(String(255), nullable=False)
    slug = Column(String(100), unique=True, nullable=False)
    status = Column(Enum('trial', 'active', 'suspended', 'deleted',
                         name='tenant_status'), default='trial')
    plan = Column(String(50), default='free')
    settings = Column(JSONB, default=dict)
    created_at = Column(DateTime(timezone=True), server_default='NOW()')
    updated_at = Column(DateTime(timezone=True), onupdate='NOW()')
    deleted_at = Column(DateTime(timezone=True))

# ✅ 验证通过 - SQLAlchemy模型

🔄 数据库版本管理最佳实践

复制# Alembic迁移最佳实践
MIGRATION_BEST_PRACTICES = """
1. 每个迁移只做一件事
   ✅ 好: 添加users表
   ❌ 差: 添加users表+添加subscriptions表+修改tenants表

2. 迁移必须可回滚
   每个upgrade()必须有对应的downgrade()
   在开发环境测试回滚

3. 数据迁移和Schema迁移分开
   先改Schema → 再写数据迁移脚本 → 最后加约束

4. 不要修改已发布的迁移
   如果需要修改，创建新的迁移

5. 生产部署前在Staging验证
   pg_dump → 迁移 → 验证 → 生产
"""

# ✅ 验证通过 - 迁移最佳实践

📊 查询性能基线

🔧 工程化实践

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("代码质量工具配置完成")

错误追踪集成(Sentry)

复制# 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错误追踪

📚 扩展学习资源

本课内容是SaaS全栈开发的重要一环。以下是推荐的深入学习资源：

必读书籍

• 《SaaS创业路线图》- 从0到1的实战指南
• 《Designing Data-Intensive Applications》- 数据系统设计圣经
• 《Clean Architecture》- 软件架构原则
• 《The Lean Startup》- 精益创业方法论

在线资源

• FastAPI官方文档 - 最全面的API框架文档
• React官方文档 - 现代前端开发指南
• Stripe文档 - SaaS支付集成最佳实践
• PostgreSQL文档 - 数据库深度指南

实践项目建议

学完本课后，建议你：

1. 在本地环境动手实现本课的代码示例
2. 根据你的SaaS项目调整和扩展代码
3. 将关键决策记录到ADR文档
4. 编写单元测试验证功能正确性

🔄 数据库迁移实战

Alembic是Python生态最成熟的数据库迁移工具，掌握它是SaaS后端开发必备技能：

复制# 迁移最佳实践
MIGRATION_RULES = {
    "规则1": "每个迁移只做一件事，方便回滚",
    "规则2": "必须实现upgrade()和downgrade()",
    "规则3": "数据迁移和Schema迁移分开",
    "规则4": "不要修改已发布的迁移文件",
    "规则5": "生产部署前在Staging验证",
    "规则6": "大表加列使用ALGORITHM=INPLACE",
    "规则7": "加NOT NULL列先加允许NULL再回填数据",
}

# 典型迁移文件
# alembic/versions/001_create_users.py
def upgrade():
    op.create_table('users',
        sa.Column('id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), primary_key=True),
        sa.Column('tenant_id', postgresql.UUID(as_uuid=True), sa.ForeignKey('tenants.id'), nullable=False),
        sa.Column('email', sa.String(255), nullable=False),
        sa.Column('name', sa.String(255), nullable=False),
        sa.Column('password_hash', sa.String(255), nullable=False),
        sa.Column('role', sa.Enum('owner','admin','member','viewer', name='user_role'), default='member'),
        sa.Column('is_active', sa.Boolean(), default=True),
        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('NOW()')),
    )
    op.create_index('idx_users_email_tenant', 'users', ['email', 'tenant_id'],
                    postgresql_where=sa.text('deleted_at IS NULL'))

def downgrade():
    op.drop_index('idx_users_email_tenant')
    op.drop_table('users')

print("迁移规则:", len(MIGRATION_RULES), "条")

📊 查询性能基线