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

Python 全栈开发：MySQL 与 Redis 数据库集成指南 | 极客日志

import sqlite3

# 1. 连接到数据库（如果不存在则创建）
conn = sqlite3.connect("my_app.db")
cursor = conn.cursor()

# 2. 创建表
cursor.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE
    )
""")

# 3. 插入数据（注意：永远不要用 f-string 拼 SQL，会有注入风险！）
cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", ("张三", "[email protected]"))

# 4. 提交并关闭
conn.commit()
conn.close()

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

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    age: int

# 创建引擎（这里用 SQLite）
engine = create_engine("sqlite:///database.db")
SQLModel.metadata.create_all(engine)

# 插入数据
with Session(engine) as session:
    user1 = User(name="李四", age=25)
    session.add(user1)
    session.commit()

# 查询数据
with Session(engine) as session:
    statement = select(User).where(User.name == "李四")
    results = session.exec(statement)
    for user in results:
        print(f"找到用户：{user.name}, 年龄：{user.age}")

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()  # 关闭会话，连接归还到池中

阶段	说明
初始化	应用启动时，创建 pool_size 个连接
请求连接	从池中获取可用连接
连接不足	若池已满，创建 overflow 连接（不超过 max_overflow）
连接归还	使用完毕后，连接回到池中等待复用
连接回收	超过 pool_recycle 时间的连接会被重建

特性	英文	含义	生活比喻
原子性	Atomicity	事务要么全部成功，要么全部失败	转账：扣款和加款必须同时完成
一致性	Consistency	事务前后，数据库始终处于合法状态	转账前后，两人余额总和不变
隔离性	Isolation	并发事务互不干扰	两个人同时取钱，不会取到同一张钞票
持久性	Durability	事务提交后，数据永久保存	转账成功后，即使断电也不会丢失

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") # 排他锁

隔离级别	脏读	不可重复读	幻读	性能
READ UNCOMMITTED	可能	可能	可能	最快
READ COMMITTED	不会	可能	可能	较快
REPEATABLE READ	不会	不会	可能	中等
SERIALIZABLE	不会	不会	不会	最慢

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

索引类型	适用场景	示例
主键索引	唯一标识每行数据	PRIMARY KEY (id)
唯一索引	确保列值不重复	UNIQUE (email)
普通索引	加速查询	INDEX (name)
组合索引	多列联合查询	INDEX (name, age)
全文索引	文本搜索	FULLTEXT (content)

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()

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

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()

关键字	含义	性能影响
SCAN TABLE	全表扫描	性能差，数据量大时避免
SEARCH TABLE	使用索引查找	性能好
USING INDEX	使用了索引	正面指标
USING COVERING INDEX	覆盖索引（无需回表）	最佳性能

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

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

优点	缺点
代码可读性强	性能开销（生成 SQL 需要时间）
数据库无关性（切换数据库方便）	复杂查询难以优化
自动处理关联关系	学习曲线陡峭
内置连接池和事务管理	隐藏了 SQL 细节，可能写出低效代码

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

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

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')}")

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")

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)

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}")

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}")

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

1. 为什么我们需要数据库？

2. SQL：像写英文一样操作数据

2.1 SQLite：零配置的实战神器

2.2 避坑指南：SQL 注入

3. ORM：像操作对象一样操作数据库

3.1 SQLAlchemy 与 SQLModel

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

4.1 数据库连接池原理

4.2 事务 ACID 特性详解

4.3 索引原理与优化

4.4 SQL 执行计划基础

4.5 ORM 工作原理

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

5.1 Redis 能干啥？

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

7. 注意事项

8. 练习题

题目 1：SQLite 事务操作

题目 2：SQLModel 关联查询

题目 3：Redis 排行榜

更多推荐文章

相关免费在线工具

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

1. 为什么我们需要数据库？

2. SQL：像写英文一样操作数据

2.1 SQLite：零配置的实战神器

2.2 避坑指南：SQL 注入

3. ORM：像操作对象一样操作数据库

3.1 SQLAlchemy 与 SQLModel

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

4.1 数据库连接池原理

4.2 事务 ACID 特性详解

4.3 索引原理与优化

4.4 SQL 执行计划基础

4.5 ORM 工作原理

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

5.1 Redis 能干啥？

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

7. 注意事项

8. 练习题

题目 1：SQLite 事务操作

题目 2：SQLModel 关联查询

题目 3：Redis 排行榜

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具