Python 全栈开发：MySQL 与 Redis 数据库集成指南

建议：用了 ORM，你就再也不用担心手抖写错 SQL 关键字了，而且代码看起来非常"Pythonic"。

4. 数据库进阶：原理与优化

4.1 数据库连接池原理

想象你在银行办业务，每次都要重新排队取号、填表、等待窗口叫号——这效率太低了。数据库连接也是同样的道理。

什么是连接池？

连接池就像银行的"VIP 通道"，预先创建好一定数量的数据库连接，放在池子里备用。当程序需要操作数据库时，直接从池子里取一个连接用完归还，而不是每次都新建连接。

from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

# 创建带连接池的引擎
engine = create_engine(
    "sqlite:///database.db",
    pool_size=5,          # 连接池大小：保持 5 个连接
    max_overflow=10,      # 最大溢出连接数
    pool_timeout=30,      # 获取连接的超时时间（秒）
    pool_recycle=3600,    # 连接回收时间（秒），防止连接过期
    pool_pre_ping=True    # 使用前检查连接是否有效
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 使用连接
session = SessionLocal()
try:
    result = session.execute(text("SELECT 1"))
    print(result.fetchone())
finally:
    session.close()  # 关闭会话，连接归还到池中

连接池的工作流程：

一句话总结：连接池是"以空间换时间"的经典策略，用内存资源换取连接创建的开销。

4.2 事务 ACID 特性详解

事务是数据库的基石，ACID 是衡量事务可靠性的四个标准。

ACID 四要素：

隔离级别与并发问题：

import sqlite3

# SQLite 支持四种隔离级别
conn = sqlite3.connect("test.db", isolation_level=None)   # 自动提交模式
conn = sqlite3.connect("test.db", isolation_level="DEFERRED")  # 延迟锁
conn = sqlite3.connect("test.db", isolation_level="IMMEDIATE") # 立即锁
conn = sqlite3.connect("test.db", isolation_level="EXCLUSIVE") # 排他锁

并发问题解释：

• 脏读：读到其他事务未提交的数据（可能回滚）
• 不可重复读：同一事务内，两次读取同一数据结果不同
• 幻读：同一事务内，两次查询返回的行数不同

一句话总结：隔离级别越高，数据一致性越好，但并发性能越差。根据业务场景选择合适的级别。

4.3 索引原理与优化

没有索引的数据库就像没有目录的字典，查找数据只能逐页翻。

索引是什么？

索引是一种数据结构（通常是 B+ 树），它存储了列值到行位置的映射，让查询可以跳过大部分数据直接定位。

B+ 树索引的工作原理：

 [50]
 /   \
[20, 30] [70, 90]
/ | \     / | \
[10] [25] [40] [60] [80] [100]
叶子节点存储实际数据地址

索引的类型：

索引使用示例：

import sqlite3
conn = sqlite3.connect("test.db")
cursor = conn.cursor()

# 创建表时添加索引
cursor.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        username TEXT NOT NULL,
        email TEXT UNIQUE,
        age INTEGER,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

# 手动创建普通索引
cursor.execute("CREATE INDEX IF NOT EXISTS idx_age ON users(age)")

# 创建组合索引（最左前缀原则）
cursor.execute("CREATE INDEX IF NOT EXISTS idx_name_age ON users(username, age)")

conn.commit()
conn.close()

索引优化原则：

1. WHERE 子句中的列应该建索引
2. 最左前缀原则：组合索引 (a, b, c) 可以支持查询 a、ab、abc，但不支持 b、bc
3. 避免过度索引：每个索引都会增加写入开销
4. 选择性高的列更适合建索引（如身份证号 > 性别）

查看查询是否使用索引：

-- SQLite 中使用 EXPLAIN QUERY PLAN
EXPLAIN QUERY PLAN SELECT * FROM users WHERE age > 20
-- 输出中的 "USING INDEX" 表示使用了索引

一句话总结：索引是查询的加速器，但也是写入的减速带。只在必要的列上建索引，定期清理无用索引。

4.4 SQL 执行计划基础

执行计划是数据库优化器的"作战地图"，它展示了 SQL 语句的执行步骤和成本估算。

如何查看执行计划：

import sqlite3
conn = sqlite3.connect(":memory:")
cursor = conn.cursor()

# 创建测试表
cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)")
cursor.execute("CREATE INDEX idx_age ON users(age)")

# 插入测试数据
cursor.executemany("INSERT INTO users (name, age) VALUES (?, ?)", [
    ("Alice", 25),
    ("Bob", 30),
    ("Charlie", 35)
])

# 查看执行计划
print("--- 全表扫描 ---")
cursor.execute("EXPLAIN QUERY PLAN SELECT * FROM users")
for row in cursor.fetchall():
    print(row)

print("\n--- 使用索引扫描 ---")
cursor.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE age > 25")
for row in cursor.fetchall():
    print(row)

conn.close()

执行计划中的关键信息：

优化建议：

• 避免 SELECT \*,只查询需要的列
• 避免在索引列上使用函数或类型转换
• 大数据量分页使用游标或延迟关联

一句话总结：执行计划是 SQL 优化的"体检报告"，学会读懂它，就能找到性能瓶颈。

4.5 ORM 工作原理

ORM 不是魔法，它只是把繁琐的 SQL 操作封装成了优雅的 Python 代码。

ORM 的核心工作流程：

Python 对象操作 -> ORM 引擎 -> SQL 生成器 -> 执行 SQL -> 结果映射回对象

SQLAlchemy 的架构层次：

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

# 第一层：引擎层（连接管理）
engine = create_engine("sqlite:///test.db")

# 第二层：元数据层（表结构定义）
Base = declarative_base()
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

# 第三层：会话层（事务管理）
Session = sessionmaker(bind=engine)
session = Session()

# ORM 操作会被转换为 SQL
user = session.query(User).filter(User.age > 18).first()
# 实际执行的 SQL: SELECT * FROM users WHERE age > 18 LIMIT 1

ORM 的优缺点：

延迟加载 vs 立即加载：

# 延迟加载（默认）：访问关联属性时才查询
user = session.query(User).first()
# 只查 users 表
print(user.posts)  # 此时才查询 posts 表

# 立即加载：一次性查询所有数据
from sqlalchemy.orm import joinedload
user = session.query(User).options(joinedload(User.posts)).first()

一句话总结：ORM 是效率与性能之间的平衡，简单操作用 ORM，复杂查询用原生 SQL。

5. Redis：追求极致的"快"

如果说 MySQL 是"厚重的铁柜子"，那 Redis 就是"手边的便签纸"。它把数据存在内存里，速度比 MySQL 快几百倍。

5.1 Redis 能干啥？

• 缓存：把经常访问的数据存起来，别每次都去翻慢吞吞的硬盘。
• 计数器：比如文章的阅读量、点赞数。
• 验证码过期：存一个验证码，设置 5 分钟后自动消失。

import redis

# 连接 Redis（前提是你装了 redis-server）
r = redis.Redis(host='localhost', port=6379, db=0)

# 存一个键值对，有效期 60 秒
r.set('session_token', 'abc123456', ex=60)

# 取出来
token = r.get('session_token')
print(f"Token: {token.decode('utf-8')}")

6. 综合实战：构建用户注册系统的后端逻辑

结合 SQLModel，写一个简单的用户注册和登录验证的逻辑。

from sqlmodel import SQLModel, Field, create_engine, Session, select
import hashlib

# 1. 定义数据模型
class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    username: str = Field(index=True, unique=True)
    password_hash: str

# 2. 数据库工具
engine = create_engine("sqlite:///users.db")
SQLModel.metadata.create_all(engine)

def register(username, password):
    # 密码不能存明文！这是常识。
    pwd_hash = hashlib.sha256(password.encode()).hexdigest()
    with Session(engine) as session:
        new_user = User(username=username, password_hash=pwd_hash)
        try:
            session.add(new_user)
            session.commit()
            print(f"用户 {username} 注册成功！")
        except Exception:
            print("注册失败：用户名可能已存在。")

def login(username, password):
    pwd_hash = hashlib.sha256(password.encode()).hexdigest()
    with Session(engine) as session:
        statement = select(User).where(User.username == username)
        user = session.exec(statement).first()
        if user and user.password_hash == pwd_hash:
            print("登录成功！欢迎回来。")
            return True
    print("登录失败：用户名或密码错误。")
    return False

# 测试一下
register("admin", "123456")
login("admin", "123456")

7. 注意事项

1. 密码永远别存明文：如果你的数据库泄露了，明文密码会让你的用户想掐死你。用 hashlib 或者专门的 bcrypt 加密。
2. 连接池：在高并发项目里，频繁地打开、关闭数据库连接非常慢。要学会使用 create_engine 的连接池功能。
3. 数据备份：代码丢了可以重写，数据丢了公司可能就破产了。养成定期导出 SQL 备份的好习惯。

8. 练习题

题目 1：SQLite 事务操作

需求： 使用 sqlite3 模拟转账操作。

1. 创建 accounts 表，包含 name 和 balance。
2. 插入 A (1000 元) 和 B (500 元)。
3. 使用事务：从 A 扣除 200，给 B 增加 200。
4. 如果中途报错（模拟异常），则回滚事务，保证钱不丢。

import sqlite3

def transfer(from_user, to_user, amount):
    conn = sqlite3.connect(":memory:")  # 使用内存数据库方便测试
    cursor = conn.cursor()

    # 初始化数据
    cursor.execute("CREATE TABLE accounts (name TEXT, balance REAL)")
    cursor.executemany("INSERT INTO accounts VALUES (?, ?)", [
        ("A", 1000),
        ("B", 500)
    ])
    conn.commit()

    try:
        print("--- 开始转账 ---")
        # 开启事务（默认开启，只要没 commit 就在事务中）
        # 1. 扣钱
        cursor.execute("UPDATE accounts SET balance = balance - ? WHERE name = ?", (amount, from_user))
        # 模拟异常：比如银行系统突然断电
        # raise Exception("系统崩溃！")
        # 2. 加钱
        cursor.execute("UPDATE accounts SET balance = balance + ? WHERE name = ?", (amount, to_user))
        conn.commit()  # 只有两步都成功才提交
        print("转账成功！")
    except Exception as e:
        conn.rollback()  # 出错了就回滚，当无事发生
        print(f"转账失败，已回滚：{e}")
    finally:
        # 查询最终结果
        cursor.execute("SELECT * FROM accounts")
        print(cursor.fetchall())
        conn.close()

transfer("A", "B", 200)

题目 2：SQLModel 关联查询

需求： 定义两个模型：Team（球队）和 Player（球员）。 一个球队有多个球员（一对多）。

1. 创建 'Lakers' 球队。
2. 创建 'James', 'Davis' 属于 Lakers。
3. 查询 Lakers 队的所有球员名字。

from typing import List, Optional
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    players: List["Player"] = Relationship(back_populates="team")

class Player(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    team_id: Optional[int] = Field(default=None, foreign_key="team.id")
    team: Optional[Team] = Relationship(back_populates="players")

# 初始化数据库
engine = create_engine("sqlite:///")
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    # 创建数据
    lakers = Team(name="Lakers")
    p1 = Player(name="James", team=lakers)
    p2 = Player(name="Davis", team=lakers)
    session.add(lakers)
    # 级联添加
    session.commit()
    session.refresh(lakers)

    # 查询
    print(f"球队：{lakers.name}")
    for p in lakers.players:
        print(f"球员：{p.name}")

题目 3：Redis 排行榜

需求： 模拟一个简单的游戏排行榜。

1. 使用 Redis 的 zadd 添加玩家分数：player1: 100, player2: 200, player3: 150。
2. 使用 zrevrange 获取前 3 名玩家及其分数（从高到低）。

import redis

# 确保你有 Redis 服务在运行
try:
    r = redis.Redis(host='localhost', port=6379, db=0)
    # 清空旧数据
    r.delete('game_score')
    
    # 1. 添加分数 (mapping={name: score})
    r.zadd('game_score', {'player1': 100, 'player2': 200, 'player3': 150})
    
    # 2. 获取前三名 (withscores=True 同时返回分数)
    # zrevrange 是从大到小排序，0 -1 表示取全部
    top_players = r.zrevrange('game_score', 0, 2, withscores=True)
    print("--- 排行榜 ---")
    for rank, (name, score) in enumerate(top_players, 1):
        print(f"第 {rank} 名：{name.decode()} - {int(score)}分")
except Exception as e:
    print(f"Redis 连接失败，请检查服务是否启动：{e}")