上一篇我们跑通了第一个 FastAPI 接口，但数据只存在内存里 —— 刷新服务，/app/hello返回的 “Hello 小明” 就没了。这篇我们要解决 **“数据持久化”** 问题：用 ORM（SQLAlchemy）把 Python 类和 MySQL 表关联起来，让用户、角色、权限这些数据 “落地” 到数据库，刷新服务也不会丢。

一、先搞懂：ORM 到底是啥？

小白最常问：“我直接写 SQL 不行吗？为什么要用 ORM？”—— 举个例子你就懂：

• 用 SQL 插入用户：要写INSERT INTO users (username, password, email) VALUES (?, ?, ?)，还要处理连接、提交、关闭；
• 用 ORM 插入用户：创建User实例user = User(username="小明", password="123456", email="xiaoming@test.com")，调用db.add(user)+db.commit()就行，不用写 SQL。

ORM 的核心价值：把 “数据库表” 映射成 “Python 类”，把 “SQL 操作” 变成 “对象操作”，减少重复代码，避免 SQL 注入，让你专注业务逻辑。

SQLAlchemy 的 4 个核心概念（必懂！）

SQLAlchemy 是 Python 生态最强大的 ORM 框架，我们用它来关联 FastAPI 和 MySQL。先搞懂这 4 个概念：

1. Engine：数据库连接池（像 “餐厅的厨房”）—— 管理数据库连接，避免每次操作都新建连接（费资源）。
2. Session：操作数据库的 “把手”（像 “服务员”）—— 用它执行增删改查，用完要关闭。
3. Base：模型基类（像 “菜单模板”）—— 所有数据库表对应的 Python 类都要继承它，它负责把类映射到表。
4. Model：具体的表类（像 “菜单上的菜”）—— 比如User类对应users表，Role类对应roles表。

二、数据库设计：权限系统的 “核心表结构”

我们要做企业级权限管理系统，核心是 “用户 - 角色 - 权限” 的三层模型：

• 用户：系统的使用者（比如 “小明”）；
• 角色：一组权限的集合（比如 “管理员” 有 “增删用户”“查看权限” 的权限）；
• 权限：具体的操作许可（比如 “user:create” 表示能创建用户，“role:delete” 表示能删除角色）；
• 关联表：用户和角色是 “多对多”（一个用户可以有多个角色，一个角色可以给多个用户），角色和权限也是 “多对多”—— 所以需要两张关联表（user_role、role_permission）。

具体表结构设计（文字版 “ER 图”）

1. 用户表（users）：存储用户基本信息

   • id：主键（自增）
   • username：用户名（唯一，比如 “xiaoming”）
   • password_hash：密码哈希（不是明文！用 passlib 加密）
   • email：邮箱（唯一）
   • is_active：是否激活（逻辑删除，默认 True）
   • created_at：创建时间（自动填充当前时间）

2. 角色表（roles）：存储角色信息

   • id：主键
   • name：角色名（唯一，比如 “admin”）
   • description：角色描述（比如 “系统管理员”）

3. 权限表（permissions）：存储权限信息

   • id：主键
   • code：权限标识（唯一，比如 “user:create”）
   • description：权限描述（比如 “创建用户”）

4. 用户角色关联表（user_role）：关联用户和角色

   • user_id：外键（关联users.id）
   • role_id：外键（关联roles.id）
   • 联合主键（user_id+role_id，避免重复关联）

5. 角色权限关联表（role_permission）：关联角色和权限

   • role_id：外键（关联roles.id）
   • permission_id：外键（关联permissions.id）
   • 联合主键（role_id+permission_id，避免重复关联）

三、代码实现：用 SQLAlchemy 映射模型

现在把上面的表结构写成 Python 类，关联到 MySQL。

1. 配置数据库连接（从.env 读取配置）

先在项目根目录的.env文件里写数据库配置（避免硬编码密码）：

# .env文件
DATABASE_URL=mysql+pymysql://root:123456@localhost:3306/fastapi_perm_db
# 解释：mysql+pymysql://用户名:密码@主机:端口/数据库名
# 注意：要先在MySQL里创建数据库fastapi_perm_db（用Navicat或命令行`CREATE DATABASE fastapi_perm_db;`）

然后写app/config.py读取配置：

# app/config.py
from pydantic_settings import BaseSettings  # 注意：Pydantic v2用BaseSettings，v1用BaseSettings从pydantic import BaseSettings

class Settings(BaseSettings):
    # 数据库配置
    DATABASE_URL: str

    class Config:
        env_file = ".env"  # 指定.env文件路径
        env_file_encoding = "utf-8"

# 创建配置实例（全局用这个实例取配置）
settings = Settings()

2. 初始化 SQLAlchemy（创建 Engine、Base、Session）

写app/db.py，负责 SQLAlchemy 的初始化：

# app/db.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.config import settings

# 1. 创建Engine（连接池）
engine = create_engine(
    settings.DATABASE_URL,
    echo=True,  # 开发模式下打印SQL语句（方便调试），上线要关掉
    pool_size=5,  # 连接池大小（默认5）
    max_overflow=10  # 超出连接池后的最大连接数（默认10）
)

# 2. 创建Session工厂（用它生成Session实例）
SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,  # 关闭自动刷新（避免意外提交）
    autocommit=False, # 关闭自动提交（手动控制事务）
    expire_on_commit=False  # 提交后不失效（避免查询已提交的数据报错）
)

# 3. 创建Base基类（所有模型类继承它）
Base = declarative_base()

# 4. 依赖函数：获取数据库Session（FastAPI用依赖注入）
def get_db():
    db = SessionLocal()
    try:
        yield db  # 生成Session，供接口使用
    finally:
        db.close()  # 不管有没有错误，都关闭Session（避免内存泄漏）

3. 定义模型类（映射到数据库表）

现在写各个表对应的 Python 类，放在app/models文件夹下：

（1）用户模型（app/models/user.py）

# app/models/user.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship
from app.db import Base

class User(Base):
    __tablename__ = "users"  # 对应数据库的users表（必须小写）

    # 字段定义
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True, nullable=False)  # 用户名（唯一，非空）
    password_hash = Column(String(255), nullable=False)  # 密码哈希（非空）
    email = Column(String(100), unique=True, index=True, nullable=False)  # 邮箱（唯一，非空）
    is_active = Column(Boolean, default=True)  # 是否激活（默认True）
    created_at = Column(DateTime(timezone=True), default=func.now())  # 创建时间（自动填充当前时间）

    # 关系映射：用户-角色（多对多），通过user_role表关联
    roles = relationship("Role", secondary="user_role", back_populates="users")

（2）角色模型（app/models/role.py）

# app/models/role.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship
from app.db import Base

class Role(Base):
    __tablename__ = "roles"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), unique=True, index=True, nullable=False)  # 角色名（唯一）
    description = Column(String(255))  # 角色描述

    # 关系映射：角色-用户（多对多）
    users = relationship("User", secondary="user_role", back_populates="roles")
    # 关系映射：角色-权限（多对多），通过role_permission表关联
    permissions = relationship("Permission", secondary="role_permission", back_populates="roles")

（3）权限模型（app/models/permission.py）

# app/models/permission.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship
from app.db import Base

class Permission(Base):
    __tablename__ = "permissions"

    id = Column(Integer, primary_key=True, index=True)
    code = Column(String(50), unique=True, index=True, nullable=False)  # 权限标识（唯一，比如"user:create"）
    description = Column(String(255))  # 权限描述（比如"创建用户"）

    # 关系映射：权限-角色（多对多）
    roles = relationship("Role", secondary="role_permission", back_populates="permissions")

（4）关联表（app/models/association.py）

关联表不需要写类，直接用Table定义（因为关联表没有额外字段，只有两个外键）：

# app/models/association.py
from sqlalchemy import Table, Column, Integer, ForeignKey
from app.db import Base

# 用户-角色关联表（user_role）
user_role = Table(
    "user_role",
    Base.metadata,
    Column("user_id", Integer, ForeignKey("users.id"), primary_key=True),  # 外键：关联users.id
    Column("role_id", Integer, ForeignKey("roles.id"), primary_key=True),  # 外键：关联roles.id
)

# 角色-权限关联表（role_permission）
role_permission = Table(
    "role_permission",
    Base.metadata,
    Column("role_id", Integer, ForeignKey("roles.id"), primary_key=True),
    Column("permission_id", Integer, ForeignKey("permissions.id"), primary_key=True),
)

4. 创建数据库表

现在把模型映射到数据库，创建表结构。写app/create_tables.py：

# app/create_tables.py
from app.db import Base, engine

# 创建所有继承Base的模型对应的表（第一次运行时执行）
def create_tables():
    Base.metadata.create_all(bind=engine)
    print("数据库表创建成功！")

if __name__ == "__main__":
    create_tables()

运行它：打开 cmd，进入项目根目录，输入：

python -m app.create_tables

验证：用 Navicat 连接 MySQL，打开fastapi_perm_db数据库，能看到users、roles、permissions、user_role、role_permission这五张表 —— 说明模型映射成功！

三、测试：用 ORM 操作数据库（增删改查）

现在写个测试脚本，插入用户、角色、权限，然后查询关联数据，验证是否正确。

1. 写测试脚本（app/tests/test_db.py）

# app/tests/test_db.py
from sqlalchemy.orm import Session
from app.db import SessionLocal, engine, get_db
from app.models.user import User
from app.models.role import Role
from app.models.permission import Permission
from passlib.context import CryptContext  # 密码哈希库

# 初始化密码哈希器（用bcrypt算法）
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def test_db_operations():
    # 1. 获取Session
    db = SessionLocal()

    try:
        # ---------------------- 插入测试数据 ----------------------
        # a. 创建权限（比如"user:create"和"role:delete"）
        perm1 = Permission(code="user:create", description="创建用户")
        perm2 = Permission(code="role:delete", description="删除角色")
        db.add_all([perm1, perm2])
        db.commit()  # 提交到数据库

        # b. 创建角色（比如"admin"，关联上面的两个权限）
        admin_role = Role(name="admin", description="系统管理员")
        admin_role.permissions.append(perm1)
        admin_role.permissions.append(perm2)
        db.add(admin_role)
        db.commit()

        # c. 创建用户（比如"xiaoming"，密码哈希是"123456"）
        user = User(
            username="xiaoming",
            password_hash=pwd_context.hash("123456"),  # 加密密码
            email="xiaoming@test.com"
        )
        user.roles.append(admin_role)  # 给用户分配admin角色
        db.add(user)
        db.commit()

        # ---------------------- 查询测试数据 ----------------------
        # a. 查询用户"xiaoming"
        db_user = db.query(User).filter(User.username == "xiaoming").first()
        print(f"查询到用户：{db_user.username}，邮箱：{db_user.email}")

        # b. 查询用户的角色
        print(f"用户的角色：{[role.name for role in db_user.roles]}")  # 应该输出["admin"]

        # c. 查询角色的权限
        print(f"admin角色的权限：{[perm.code for perm in db_user.roles[0].permissions]}")  # 应该输出["user:create", "role:delete"]

    finally:
        # 不管有没有错误，都关闭Session
        db.close()

if __name__ == "__main__":
    test_db_operations()

2. 运行测试脚本

打开 cmd，输入：

python -m app.tests.test_db

输出结果（类似）：

查询到用户：xiaoming，邮箱：xiaoming@test.com
用户的角色：['admin']
admin角色的权限：['user:create', 'role:delete']

验证数据库：用 Navicat 查看users表，能看到xiaoming的记录；roles表有admin；permissions表有user:create和role:delete；user_role表有user_id（xiaoming 的 id）和role_id（admin 的 id）；role_permission表有role_id（admin 的 id）和permission_id（两个权限的 id）—— 完美！

四、常见问题排查（小白踩坑高频区）

1. 运行 create_tables.py 报错 “Unknown database 'fastapi_perm_db'”

• 原因：没在 MySQL 里创建数据库fastapi_perm_db；
• 解决：用 Navicat 或命令行创建：CREATE DATABASE fastapi_perm_db;。

2. 插入数据报错 “Duplicate entry 'xiaoming' for key 'users.username'”

• 原因：用户名重复（username字段设了unique=True）；
• 解决：换个用户名（比如 “xiaohong”），或删除数据库里已有的 “xiaoming” 记录。

3. 查询角色权限时报错 “AttributeError: 'Role' object has no attribute 'permissions'”

• 原因：Role类里没定义permissions关系映射；
• 解决：检查Role类的permissions = relationship(...)代码是否正确。

五、总结：这篇我们做了什么？

1. 搞懂了 SQLAlchemy 的核心概念（Engine、Session、Base、Model）；
2. 设计了权限系统的核心表结构（用户 - 角色 - 权限 - 关联表）；
3. 用 SQLAlchemy 实现了模型类，把 Python 类映射到 MySQL 表；
4. 插入了测试数据，验证了 “用户 - 角色 - 权限” 的关联关系；
5. 学会了用 ORM 操作数据库（不用写 SQL！）。