SQLAlchemy快速入门自我总结

SQLAlchemy 精通教程

之前写了一个sqlalchemy的简单实践，这次从入门到精通，自我总结，可能有缺，持续更新。。。

---

目录

1. SQLAlchemy 简介
2. 核心概念一图流
3. 快速开始
4. 声明式模型
5. 核心概念详解
6. CRUD 操作
7. 关系映射
8. 查询高级用法
9. 会话管理
10. 事务与锁
11. 异步支持 (2.0+)
12. 性能优化
13. 迁移管理 (Alembic)
14. 最佳实践

---

1. SQLAlchemy 简介

什么是 SQLAlchemy？

SQLAlchemy 是 Python 中最著名的 SQL 工具包和 ORM（对象关系映射）框架。

┌─────────────────────────────────────────────────────────┐
│                    SQLAlchemy 架构                        │
├─────────────────────────────────────────────────────────┤
│                                                         │
│   应用层：你的 Python 代码                                  │
│         ↓                                               │
│   ORM 层：将 Python 类映射到数据库表                         │
│         ↓                                               │
│   Core 层：SQL 表达式语言                                 │
│         ↓                                               │
│   方言层：适配不同数据库 (MySQL, PostgreSQL, SQLite...)   │
│         ↓                                               │
│   DBAPI：数据库驱动 (pymysql, psycopg2, aiomysql...)     │
│                                                         │
└─────────────────────────────────────────────────────────┘

SQLAlchemy 1.x vs 2.0

特性	1.x	2.0
异步支持	实验性	完整支持
类型提示	有限	完整支持
API 风格	混合	更清晰分离
推荐用法	session.query()	select()

安装

# 基础安装
pip install sqlalchemy

# MySQL 支持
pip install pymysql
pip install sqlalchemy[mysql]

# PostgreSQL 支持
pip install psycopg2-binary
pip install sqlalchemy[postgresql]

# 异步支持
pip install aiomysql      # MySQL 异步
pip install asyncpg       # PostgreSQL 异步

---

2. 核心概念一图流

┌────────────────────────────────────────────────────────────────┐
│                     SQLAlchemy 核心概念                          │
├────────────────────────────────────────────────────────────────┤
│                                                                │
│  Engine（引擎）                                                 │
│  └── 数据库连接的工厂，管理连接池                               │
│      └── URL: "dialect+driver://user:pass@host/db"             │
│                                                                │
│  Metadata（元数据）                                             │
│  └── 存储表结构信息，用于创建/删除表                            │
│                                                                │
│  Table（表） - Core 层                                          │
│  └── 定义数据库表结构（列、约束等）                              │
│                                                                │
│  Model（模型） - ORM 层                                         │
│  └── Python 类，映射到数据库表                                   │
│      └── 属性 → 列，实例 → 行                                   │
│                                                                │
│  Session（会话）                                                │
│  └── 与数据库的"对话"，负责 CRUD 操作                            │
│      └── 类似"暂存区"，commit 后才写入数据库                      │
│                                                                │
│  Query / select()                                               │
│  └── 构建 SQL 查询                                               │
│                                                                │
└────────────────────────────────────────────────────────────────┘

典型工作流程

1. 创建 Engine → 2. 创建 Session → 3. 定义模型
                                        ↓
4. 使用 Session 执行查询/操作 → 5. 提交事务 → 6. 关闭 Session

---

3. 快速开始

最小可运行示例

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

# 1. 创建引擎（数据库连接）
engine = create_engine("sqlite:///example.db", echo=True)

# 2. 创建基类
Base = declarative_base()

# 3. 定义模型
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

# 4. 创建表
Base.metadata.create_all(engine)

# 5. 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 6. 添加数据
user = User(name="张三", email="[email protected]")
session.add(user)
session.commit()

# 7. 查询数据
users = session.query(User).all()
for u in users:
    print(f"{u.id}: {u.name} - {u.email}")

# 8. 关闭会话
session.close()

连接字符串详解

from sqlalchemy import create_engine

# SQLite（文件）
engine = create_engine("sqlite:///example.db")

# SQLite（内存）
engine = create_engine("sqlite:///:memory:")

# MySQL
engine = create_engine(
    "mysql+pymysql://user:password@localhost:3306/dbname?charset=utf8mb4"
)

# PostgreSQL
engine = create_engine(
    "postgresql+psycopg2://user:password@localhost:5432/dbname"
)

# PostgreSQL（异步）
engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/dbname"
)

# 带连接池配置
engine = create_engine(
    "sqlite:///example.db",
    pool_size=10,           # 连接池大小
    max_overflow=20,        # 超出 pool_size 后允许的最大连接数
    pool_pre_ping=True,     # 使用前检查连接是否有效
    pool_recycle=3600,      # 连接回收时间（秒）
)

---

4. 声明式模型

基础模型定义

from sqlalchemy import Column, Integer, String, Float, DateTime, Boolean, Text
from sqlalchemy.orm import declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"  # 表名
    
    # 主键
    id = Column(Integer, primary_key=True, autoincrement=True)
    
    # 普通字段
    username = Column(String(50), nullable=False, unique=True)
    email = Column(String(100), nullable=False)
    password_hash = Column(String(255), nullable=False)
    
    # 带默认值
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # 大文本
    bio = Column(Text)
    
    # 数值类型
    age = Column(Integer)
    balance = Column(Float, default=0.0)
    
    def __repr__(self):
        return f"<User(id={self.id}, username={self.username})>"

常用字段类型

from sqlalchemy import (
    # 字符串类型
    String, Text, Unicode,
    
    # 整数类型
    Integer, SmallInteger, BigInteger,
    
    # 浮点类型
    Float, Numeric, Decimal,
    
    # 日期时间
    Date, Time, DateTime,
    
    # 布尔
    Boolean,
    
    # 二进制
    LargeBinary, PickleType,
    
    # JSON (PostgreSQL/MySQL 5.7+)
    JSON,
    
    # UUID
    UUID,
    
    # 其他
    Enum, Interval,
)

class AllTypes(Base):
    __tablename__ = "all_types"
    
    id = Column(Integer, primary_key=True)
    
    # 字符串
    short_text = Column(String(50))      # VARCHAR(50)
    long_text = Column(Text)             # TEXT
    
    # 整数
    small_num = Column(SmallInteger)     # SMALLINT
    normal_num = Column(Integer)         # INTEGER
    big_num = Column(BigInteger)         # BIGINT
    
    # 浮点
    float_num = Column(Float)            # FLOAT
    precise_num = Column(Numeric(10, 2)) # DECIMAL(10, 2)
    
    # 日期时间
    date_only = Column(Date)             # DATE
    time_only = Column(Time)             # TIME
    datetime_full = Column(DateTime)     # DATETIME
    
    # 布尔
    is_true = Column(Boolean)            # BOOLEAN
    
    # JSON
    data = Column(JSON)                  # JSON
    
    # 带枚举
    status = Column(Enum("active", "inactive", "banned"))

字段选项详解

from sqlalchemy import Column, Integer, String, ForeignKey, Index, UniqueConstraint

class DetailedModel(Base):
    __tablename__ = "detailed"
    
    id = Column(
        Integer,
        primary_key=True,        # 主键
        autoincrement=True,      # 自增
        nullable=False,          # 不允许 NULL
        index=True,              # 创建索引
        unique=True,             # 唯一约束
    )
    
    name = Column(
        String(50),
        nullable=False,          # 不允许 NULL
        default="Unnamed",       # 默认值（Python 层）
        server_default="Unnamed", # 默认值（数据库层）
    )
    
    email = Column(
        String(100),
        unique=True,             # 唯一约束
        index=True,              # 索引
    )
    
    category_id = Column(
        Integer,
        ForeignKey("categories.id", ondelete="CASCADE"),  # 外键 + 级联删除
        nullable=True,
    )
    
    # 复合唯一约束
    __table_args__ = (
        UniqueConstraint("name", "category_id", name="uq_name_category"),
        Index("ix_email_status", "email", "status"),  # 复合索引
        {"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"},  # MySQL 特定选项
    )

---

5. 核心概念详解

Engine（引擎）

from sqlalchemy import create_engine, text

# 创建引擎
engine = create_engine(
    "sqlite:///example.db",
    echo=True,           # 打印 SQL 语句（调试用）
    pool_pre_ping=True,  # 自动重连
)

# 直接执行原生 SQL
with engine.connect() as conn:
    result = conn.execute(text("SELECT version()"))
    print(result.fetchone())

# 事务执行
with engine.begin() as conn:
    conn.execute(text("INSERT INTO users (name) VALUES ('test')"))
    # 自动 commit，如果异常则 rollback

Metadata（元数据）

from sqlalchemy import MetaData, Table, create_engine

engine = create_engine("sqlite:///example.db")
metadata = MetaData()

# 从现有数据库反射表结构
users_table = Table("users", metadata, autoload_with=engine)

# 查看所有表
print(metadata.tables.keys())

# 创建所有表
metadata.create_all(engine)

# 删除所有表
metadata.drop_all(engine)

# 反射整个数据库
metadata.reflect(engine)

核心层 Table（不使用 ORM）

from sqlalchemy import Table, Column, Integer, String, MetaData, insert, select

metadata = MetaData()

# 定义表（Core 方式）
users = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50)),
    Column("email", String(100)),
)

# 插入数据
stmt = insert(users).values(name="张三", email="[email protected]")

with engine.begin() as conn:
    result = conn.execute(stmt)
    print(f"插入的 ID: {result.inserted_primary_key[0]}")

# 查询
stmt = select(users).where(users.c.name == "张三")
with engine.connect() as conn:
    result = conn.execute(stmt)
    for row in result:
        print(row.id, row.name, row.email)

---

6. CRUD 操作

创建 (Create)

from sqlalchemy.orm import Session

# 创建单个对象
user = User(name="李四", email="[email protected]")
session.add(user)
session.commit()
session.refresh(user)  # 刷新获取自增 ID
print(f"ID: {user.id}")

# 批量添加
users = [
    User(name="用户 1", email="[email protected]"),
    User(name="用户 2", email="[email protected]"),
    User(name="用户 3", email="[email protected]"),
]
session.add_all(users)
session.commit()

# 使用字典批量创建
data = [
    {"name": "张三", "email": "[email protected]"},
    {"name": "李四", "email": "[email protected]"},
]
session.bulk_insert_mappings(User, data)
session.commit()

查询 (Read)

from sqlalchemy import select

# 方法 1：传统 query API（1.x 风格，但仍可用）
user = session.query(User).first()           # 第一条
user = session.query(User).get(1)            # 按 ID 获取
users = session.query(User).all()            # 全部
users = session.query(User).limit(10).all()  # 限制 10 条
users = session.query(User).offset(10).all() # 跳过 10 条

# 方法 2：2.0 推荐 select() API
stmt = select(User)
user = session.execute(stmt).first()

stmt = select(User).limit(10)
users = session.execute(stmt).scalars().all()

# 条件查询
user = session.query(User).filter(User.id == 1).first()
users = session.query(User).filter(User.name == "张三").all()

# 多个条件
users = session.query(User).filter(
    User.is_active == True,
    User.age >= 18
).all()

# OR 条件
from sqlalchemy import or_
users = session.query(User).filter(
    or_(User.name == "张三", User.name == "李四")
).all()

# LIKE 查询
users = session.query(User).filter(
    User.email.like("%@gmail.com")
).all()

# IN 查询
users = session.query(User).filter(
    User.id.in_([1, 2, 3])
).all()

# 排序
users = session.query(User).order_by(User.created_at.desc()).all()
users = session.query(User).order_by(User.name.asc(), User.id.desc()).all()

# 只查询特定字段
results = session.query(User.id, User.name).all()
for id, name in results:
    print(id, name)

# 计数
count = session.query(User).count()

# 聚合
from sqlalchemy import func
max_id = session.query(func.max(User.id)).scalar()
avg_age = session.query(func.avg(User.age)).scalar()

# 分组
results = session.query(
    User.status,
    func.count(User.id)
).group_by(User.status).all()

更新 (Update)

# 方法 1：获取对象后修改
user = session.query(User).get(1)
if user:
    user.name = "新名字"
    user.email = "[email protected]"
    session.commit()

# 方法 2：bulk_update（批量更新）
session.bulk_update_mappings(
    User,
    [{"id": 1, "name": "更新 1"}, {"id": 2, "name": "更新 2"}]
)
session.commit()

# 方法 3：SQL 更新（高效，不加载对象）
from sqlalchemy import update

stmt = update(User).where(User.id == 1).values(name="新名字")
session.execute(stmt)
session.commit()

# 批量条件更新
stmt = update(User).where(User.is_active == False).values(is_active=True)
session.execute(stmt)
session.commit()

删除 (Delete)

# 方法 1：获取对象后删除
user = session.query(User).get(1)
if user:
    session.delete(user)
    session.commit()

# 方法 2：SQL 删除
from sqlalchemy import delete

stmt = delete(User).where(User.id == 1)
session.execute(stmt)
session.commit()

# 批量删除
stmt = delete(User).where(User.is_active == False)
result = session.execute(stmt)
print(f"删除了 {result.rowcount} 条记录")
session.commit()

---

7. 关系映射

一对一关系

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    
    # 关系定义
    profile = relationship("Profile", back_populates="user", uselist=False)

class Profile(Base):
    __tablename__ = "profiles"
    
    id = Column(Integer, primary_key=True)
    bio = Column(Text)
    user_id = Column(Integer, ForeignKey("users.id"), unique=True)
    
    # 反向关系
    user = relationship("User", back_populates="profile")

# 使用示例
user = User(name="张三")
user.profile = Profile(bio="这是我的简介")

session.add(user)
session.commit()

# 访问关联对象
print(user.profile.bio)
print(user.profile.user.name)  # 反向访问

一对多关系

class Parent(Base):
    __tablename__ = "parents"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    
    # 一方：可以有多个孩子
    children = relationship("Child", back_populates="parent")

class Child(Base):
    __tablename__ = "children"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    parent_id = Column(Integer, ForeignKey("parents.id"))
    
    # 多方：每个孩子有一个父亲
    parent = relationship("Parent", back_populates="children")

# 使用示例
parent = Parent(name="父亲")
parent.children = [
    Child(name="孩子 1"),
    Child(name="孩子 2"),
    Child(name="孩子 3"),
]

session.add(parent)
session.commit()

# 访问
for child in parent.children:
    print(f"{child.name} 的父亲是 {child.parent.name}")

# 查询某个父亲的所有孩子
children = session.query(Child).filter(Child.parent_id == 1).all()

多对多关系

from sqlalchemy import Table

# 关联表
user_roles = Table(
    "user_roles",
    Base.metadata,
    Column("user_id", Integer, ForeignKey("users.id")),
    Column("role_id", Integer, ForeignKey("roles.id")),
    PrimaryKeyConstraint("user_id", "role_id"),  # 复合主键
)

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    
    # 多对多关系
    roles = relationship(
        "Role",
        secondary=user_roles,
        back_populates="users"
    )

class Role(Base):
    __tablename__ = "roles"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    
    users = relationship("User", secondary=user_roles, back_populates="roles")

# 使用示例
admin_role = Role(name="管理员")
editor_role = Role(name="编辑")

user = User(name="张三")
user.roles = [admin_role, editor_role]

session.add_all([user, admin_role, editor_role])
session.commit()

# 查询用户的所有角色
user = session.query(User).filter(User.name == "张三").first()
for role in user.roles:
    print(f"角色：{role.name}")

# 查询有管理员角色的所有用户
admins = session.query(User).filter(
    User.roles.any(Role.name == "管理员")
).all()

级联操作

class Parent(Base):
    __tablename__ = "parents"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    
    # 级联配置
    children = relationship(
        "Child",
        back_populates="parent",
        cascade="all, delete-orphan",  # 删除父亲时删除所有孩子
    )

class Child(Base):
    __tablename__ = "children"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    parent_id = Column(Integer, ForeignKey("parents.id"))
    parent = relationship("Parent", back_populates="children")

# cascade 选项说明：
# "save-update" - 保存/更新时级联
# "merge" - merge 操作时级联
# "refresh-expire" - refresh/expire 时级联
# "expunge" - expunge 时级联
# "delete" - 删除时级联
# "delete-orphan" - 删除孤儿记录
# "all" - 以上所有
# "all, delete-orphan" - 常用组合

自引用关系

class Employee(Base):
    __tablename__ = "employees"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    
    # 外键指向自己
    manager_id = Column(Integer, ForeignKey("employees.id"))
    
    # 下属（一对多）
    subordinates = relationship(
        "Employee",
        back_populates="manager",
        foreign_keys=[manager_id]
    )
    
    # 上级（多对一）
    manager = relationship(
        "Employee",
        back_populates="subordinates",
        remote_side=[id]
    )

# 使用示例
ceo = Employee(name="CEO")
manager = Employee(name="经理", manager=ceo)
employee = Employee(name="员工", manager=manager)

session.add_all([ceo, manager, employee])
session.commit()

# 访问
print(ceo.subordinates[0].name)  # 经理
print(employee.manager.name)     # 经理
print(employee.manager.manager.name)  # CEO

---

8. 查询高级用法

Join 查询

from sqlalchemy.orm import joinedload, selectinload

# 内连接
results = session.query(User, Profile).join(Profile).all()

# 左外连接
results = session.query(User).outerjoin(Profile).all()

# 多个 join
results = session.query(User).join(Profile).join(Post).all()

# 使用关系自动推断连接条件
users_with_posts = session.query(User).join(User.posts).all()

# 明确指定连接条件
results = session.query(User).join(
    Profile, User.id == Profile.user_id
).all()

预加载（解决 N+1 问题）

# 问题：N+1 查询
# 执行 1 次查询获取用户，然后每个用户执行 1 次查询获取帖子
users = session.query(User).all()
for user in users:
    print(user.posts)  # 每次都触发新查询！

# 解决 1：joinedload（适合一对一、一对少）
users = session.query(User).options(
    joinedload(User.profile)
).all()

# 解决 2：selectinload（适合一对多）
users = session.query(User).options(
    selectinload(User.posts)
).all()

# 解决 3：selectinload_all（多层嵌套）
users = session.query(User).options(
    selectinload_all(User.posts, User.profile)
).all()

# 多层预加载
users = session.query(User).options(
    joinedload(User.posts).joinedload(Post.comments)
).all()

子查询

from sqlalchemy import func

# 标量子查询
subquery = session.query(func.count(Post.id)).filter(
    Post.user_id == User.id
).correlate(User).as_scalar()

users = session.query(User, subquery.label("post_count")).all()

# IN 子查询
active_user_ids = session.query(User.id).filter(User.is_active == True)
posts = session.query(Post).filter(
    Post.user_id.in_(active_user_ids)
).all()

# EXISTS 子查询
from sqlalchemy import exists

exists_query = exists().where(Post.user_id == User.id)
users_with_posts = session.query(User).filter(exists_query).all()

窗口函数

from sqlalchemy import func, over

# ROW_NUMBER
query = session.query(
    User.name,
    User.salary,
    func.row_number().over(
        partition_by=User.department,
        order_by=User.salary.desc()
    ).label("rank")
)

# 聚合窗口函数
query = session.query(
    User.name,
    User.salary,
    func.avg(User.salary).over(
        partition_by=User.department
    ).label("dept_avg")
)

原生 SQL 查询

from sqlalchemy import text

# 参数化查询（防止 SQL 注入）
stmt = text("SELECT * FROM users WHERE id = :id")
result = session.execute(stmt, {"id": 1})

# 返回 ORM 对象
stmt = text("SELECT * FROM users WHERE id = :id")
result = session.execute(stmt, {"id": 1})
user = result.scalars().first()

# 混合使用 ORM 和原生 SQL
from sqlalchemy.orm import Session

with Session(engine) as session:
    # 原生 SQL
    session.execute(text("UPDATE users SET active = 1 WHERE id = :id"), {"id": 1})
    session.commit()

---

9. 会话管理

Session 生命周期

from sqlalchemy.orm import sessionmaker, Session, scoped_session

# 创建 Session 工厂
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)

# 方法 1：手动管理
session = SessionLocal()
try:
    # 操作...
    session.commit()
except:
    session.rollback()
    raise
finally:
    session.close()

# 方法 2：上下文管理器（推荐）
with SessionLocal() as session:
    try:
        # 操作...
        session.commit()
    except:
        session.rollback()
        raise

# 方法 3：scoped_session（线程安全，适合 Web 应用）
db_session = scoped_session(SessionLocal)

# 在请求结束时清理
@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()

会话状态详解

# 对象状态转换：
# 
# transient → [add()] → pending → [commit()] → persistent
#                                ↓
#                           [rollback()]
#                                ↓
#                           detached ← [close()/expire()]

from sqlalchemy.orm import Session

session = Session()

user = User(name="test")
print(session.contains(user))  # False - transient（瞬态）

session.add(user)
print(session.contains(user))  # True - pending（待处理）

session.commit()
print(session.contains(user))  # True - persistent（持久化）

session.close()
print(session.contains(user))  # False - detached（游离）

# 重新附加
session.add(user)  # 或 session.merge(user)

会话技巧

# 批量保存（每 N 条提交一次）
def batch_insert(session, items, batch_size=1000):
    for i, item in enumerate(items):
        session.add(item)
        if (i + 1) % batch_size == 0:
            session.commit()
            session.expunge_all()  # 清空会话，释放内存
    session.commit()

# 刷新特定对象
session.refresh(user)  # 从数据库重新加载数据

# 使对象过期（下次访问时重新加载）
session.expire(user)

# 清空整个会话
session.expunge_all()

# 部分刷新（只刷新特定属性）
session.refresh(user, attribute_names=["name", "email"])

---

10. 事务与锁

事务管理

from sqlalchemy import create_engine

engine = create_engine("sqlite:///example.db")

# 方法 1：自动事务（推荐）
with engine.begin() as conn:
    conn.execute(text("INSERT INTO users (name) VALUES ('Alice')"))
    conn.execute(text("INSERT INTO users (name) VALUES ('Bob')"))
    # 如果都成功则 commit，否则自动 rollback

# 方法 2：显式事务
with engine.connect() as conn:
    trans = conn.begin()
    try:
        conn.execute(text("INSERT INTO users VALUES (1, 'Alice')"))
        conn.execute(text("INSERT INTO users VALUES (2, 'Bob')"))
        trans.commit()
    except:
        trans.rollback()
        raise

# 方法 3：嵌套事务（保存点）
with engine.begin() as conn:
    conn.execute(text("INSERT INTO users VALUES (1, 'Alice')"))
    
    with conn.begin_nested():  # 保存点
        conn.execute(text("INSERT INTO users VALUES (2, 'Bob')"))
        # 如果这里失败，只回滚到这个保存点
        # Alice 的插入仍然有效

锁机制

from sqlalchemy import select
from sqlalchemy.orm import Session

session = Session()

# 行级锁 - for_update（排他锁）
user = session.query(User).with_for_update().filter(User.id == 1).first()
# 其他事务无法修改或锁定此行

# for_update 的参数
user = session.query(User).with_for_update(
    nowait=True,    # 无法获取锁时立即抛出异常
    skip_locked=True,  # 跳过已锁定的行
    read=False,     # False=排他锁，True=共享锁
).filter(User.id == 1).first()

# 共享锁（允许其他事务读，但不允许写）
user = session.query(User).with_for_update(read=True).filter(User.id == 1).first()

# 表级锁（MySQL 示例）
session.execute(text("LOCK TABLES users WRITE"))
try:
    # 操作...
    pass
finally:
    session.execute(text("UNLOCK TABLES"))

乐观锁

from sqlalchemy import Column, Integer
from sqlalchemy.orm import relationship, Session

class Product(Base):
    __tablename__ = "products"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    stock = Column(Integer)
    version = Column(Integer, default=0, nullable=False)  # 版本号

def update_stock_optimistic(session, product_id, new_stock):
    product = session.query(Product).filter(Product.id == product_id).first()
    
    if not product:
        return False
    
    # 检查版本号是否变化
    current_version = product.version
    
    # 更新
    product.stock = new_stock
    product.version = current_version + 1
    
    try:
        session.commit()
        return True
    except:
        session.rollback()
        return False  # 版本号冲突，更新失败

---

11. 异步支持 (2.0+)

异步基础

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import select

# 异步引擎
engine = create_async_engine(
    "sqlite+aiosqlite:///example.db",
    echo=True
)

# 异步 Session
AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False
)

Base = declarative_base()

# 异步 CRUD
async def main():
    async with AsyncSessionLocal() as session:
        # 创建表
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        
        # 添加数据
        user = User(name="异步用户", email="[email protected]")
        session.add(user)
        await session.commit()
        
        # 查询
        result = await session.execute(select(User))
        users = result.scalars().all()
        
        # 获取单个
        result = await session.execute(select(User).where(User.id == 1))
        user = result.scalars().first()

异步关系预加载

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from sqlalchemy.orm import selectinload

async def get_user_with_posts(session: AsyncSession, user_id: int):
    result = await session.execute(
        select(User)
        .options(selectinload(User.posts))
        .where(User.id == user_id)
    )
    return result.scalars().first()

async def get_all_users_with_data(session: AsyncSession):
    result = await session.execute(
        select(User)
        .options(
            selectinload(User.profile),
            selectinload(User.posts).selectinload(Post.comments)
        )
    )
    return result.scalars().all()

FastAPI + 异步 SQLAlchemy

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base

app = FastAPI()

engine = create_async_engine("sqlite+aiosqlite:///example.db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession)
Base = declarative_base()

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    users = result.scalars().all()
    return [{"id": u.id, "name": u.name} for u in users]

---

12. 性能优化

批量操作

# 批量插入（高效）
def bulk_insert_users(users_data):
    # users_data = [{"name": ..., "email": ...}, ...]
    session.bulk_insert_mappings(User, users_data)
    session.commit()

# 批量更新
def bulk_update_users(updates):
    # updates = [{"id": 1, "name": "新名字"}, ...]
    session.bulk_update_mappings(User, updates)
    session.commit()

# 批量插入（使用 execute_values - PostgreSQL）
from sqlalchemy.dialects.postgresql import insert

stmt = insert(User).values(users_data)
session.execute(stmt)
session.commit()

查询优化

# 1. 只查询需要的字段
results = session.query(User.id, User.name).all()

# 2. 使用 EXISTS 代替 IN（大数据集）
from sqlalchemy import exists
users_with_posts = session.query(User).filter(
    exists().where(Post.user_id == User.id)
).all()

# 3. 使用延迟加载（适合大对象）
class Article(Base):
    content = Column(Text, lazy="dynamic")  # 访问时才查询

# 4. 禁用自动刷新
session = Session()
session.autoflush = False

# 5. 使用 yield_per 分批处理大数据集
query = session.query(User).yield_per(100)
for user in query:
    process(user)  # 每次只加载 100 条到内存

连接池优化

engine = create_engine(
    "mysql+pymysql://user:pass@localhost/db",
    
    # 连接池配置
    pool_size=20,         # 正常连接数
    max_overflow=40,      # 超出后可创建的最大连接
    pool_timeout=30,      # 等待连接的超时时间
    pool_recycle=1800,    # 连接回收时间（秒）
    pool_pre_ping=True,   # 使用前检查连接
    
    # 其他优化
    echo=False,           # 生产环境关闭 SQL 日志
)

---

13. 迁移管理 (Alembic)

Alembic 基础

# 安装
pip install alembic

# 初始化
alembic init alembic

# 配置 alembic.ini
# sqlalchemy.url = mysql+pymysql://user:pass@localhost/db

目录结构

alembic/
├── versions/          # 迁移脚本
│   ├── 001_initial.py
│   └── 002_add_user_table.py
├── env.py            # 环境配置
└── script.py.mako    # 模板

常用命令

# 生成新迁移
alembic revision --autogenerate -m "创建用户表"

# 应用迁移
alembic upgrade head

# 回退一个版本
alembic downgrade -1

# 查看当前版本
alembic current

# 查看历史
alembic history

# 生成空迁移（手动写 SQL）
alembic revision -m "添加索引"

迁移脚本示例

"""创建用户表

Revision ID: abc123
Revises: 
Create Date: 2024-01-01

"""
from alembic import op
import sqlalchemy as sa

revision = "abc123"
down_revision = None

def upgrade():
    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("username", sa.String(length=50), nullable=False),
        sa.Column("email", sa.String(length=100), nullable=False),
        sa.Column("created_at", sa.DateTime(), nullable=True),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("email"),
        sa.UniqueConstraint("username"),
    )
    op.create_index("ix_users_email", "users", ["email"], unique=False)

def downgrade():
    op.drop_index("ix_users_email", table_name="users")
    op.drop_table("users")

---

14. 最佳实践

项目结构

project/
├── app/
│   ├── __init__.py
│   ├── database.py      # 引擎、Session 配置
│   ├── models/          # 模型
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── post.py
│   └── schemas/         # Pydantic 模型
│       └── user.py
├── alembic/             # 迁移
├── tests/
└── requirements.txt

database.py 示例

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    pool_pre_ping=True,
    echo=False
)

SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
)

Base = declarative_base()

# 依赖注入用的函数
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

模型定义最佳实践

from sqlalchemy import Column, Integer, String, DateTime, func
from sqlalchemy.orm import declarative_base, Mapped, mapped_column
from datetime import datetime
from typing import Optional

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    # SQLAlchemy 2.0 风格
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow, server_default=func.now()
    )
    
    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "username": self.username,
            "email": self.email,
        }
    
    def __repr__(self) -> str:
        return f"<User(id={self.id}, username={self.username})>"

常见陷阱与避免

# ❌ 错误：忘记 commit
session.add(user)
# 应该：session.commit()

# ❌ 错误：在循环中查询
for user_id in user_ids:
    user = session.query(User).get(user_id)  # N+1 问题

# ✅ 正确：使用 IN
users = session.query(User).filter(User.id.in_(user_ids)).all()

# ❌ 错误：会话泄漏
def get_user(user_id):
    session = SessionLocal()
    return session.query(User).get(user_id)
    # 应该：用上下文管理器

# ✅ 正确：
def get_user(user_id):
    with SessionLocal() as session:
        return session.query(User).get(user_id)

# ❌ 错误：在事务中进行耗时操作
with session.begin():
    user = session.query(User).get(1)
    send_email(user.email)  # 耗时的邮件发送
    # 事务一直保持打开

# ✅ 正确：先提交，再发送
with session.begin():
    user = session.query(User).get(1)
send_email(user.email)

---

附录：快速参考

常用类型速查

String(n)       - VARCHAR(n)
Text            - TEXT
Integer         - INTEGER
Float           - FLOAT
Numeric(p, s)   - DECIMAL(p, s)
Boolean         - BOOLEAN
Date            - DATE
DateTime        - DATETIME
Time            - TIME
JSON            - JSON
LargeBinary     - BLOB
Enum(...)       - ENUM

约束速查

primary_key=True          # 主键
nullable=False            # NOT NULL
unique=True               # UNIQUE
index=True                # INDEX
default=value             # Python 默认值
server_default=text       # SQL 默认值
ForeignKey("table.col")   # 外键

关系速查

# 一对一
child = relationship("Child", uselist=False)

# 一对多
children = relationship("Child")

# 多对多
roles = relationship("Role", secondary=association_table)

# 预加载
joinedload(Model.relation)      # JOIN
selectinload(Model.relation)    # IN 查询

---

学习资源

• 官方文档：https://docs.sqlalchemy.org/
• ORM 教程：https://docs.sqlalchemy.org/en/20/orm/quickstart.html
• Alembic 文档：https://alembic.sqlalchemy.org/
• SQLAlchemy 2.0 迁移指南：https://docs.sqlalchemy.org/en/20/changelog/migration_20.html

---