一文吃透 SQLAlchemy ORM：查询、关联、事务、异步全梳理

前言

如果你用 Python 写过数据库应用，大概率听过 SQLAlchemy 的名字。它被誉为 Python 生态中最强大的 ORM（对象关系映射）工具，没有之一。但说实话，很多人对它的印象还停留在“功能强但太复杂”的阶段——尤其是面对异步、关系加载、事务管理这些进阶话题时，常常不知从何下手。

本文将带你系统梳理 SQLAlchemy ORM 的四大核心能力：查询构建、关系加载、事务管理、异步编程。读完你会明白，SQLAlchemy 2.0 已经脱胎换骨，现代 Python 的 async/await 和类型安全，它都稳稳接住了。

一、快速开始：搭建一个 Async ORM 环境

在进入正题之前，我们先搭好基础环境。以下代码展示了如何创建一个支持异步的引擎和会话工厂：

python

from sqlalchemy.ext.asyncio import AsyncAttrs, async_sessionmaker, create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, ForeignKey
from typing import List
import asyncio

# 1. 定义基类（继承 AsyncAttrs 以支持异步属性访问）
class Base(AsyncAttrs, DeclarativeBase):
pass

# 2. 创建异步引擎
engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost/db",
echo=True # 开发时打印 SQL 便于调试
)

# 3. 创建异步会话工厂
async_session = async_sessionmaker(engine, expire_on_commit=False)

expire_on_commit=False 是异步场景下的常见配置——提交事务后对象不会过期，避免下次访问时触发意外的延迟加载 。

二、模型定义：用 Python 类型描述表结构

SQLAlchemy 2.0 全面拥抱 Python 类型注解，模型定义变得非常直观：

python

class User(Base):
__tablename__ = "users"

id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
email: Mapped[str] = mapped_column(String(100), unique=True)
# 一对多关系：一个用户有多个帖子
posts: Mapped[List["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan")

class Post(Base):
__tablename__ = "posts"

id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(100))
content: Mapped[str] = mapped_column(String)
author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
# 多对一关系：一个帖子属于一个用户
author: Mapped["User"] = relationship(back_populates="posts")

使用 Mapped 类型加上 mapped_column()，既声明了 Python 类型，也声明了数据库列属性。relationship 则在两个模型之间建立了双向导航 。

三、查询：从简单筛选到复杂聚合

3.1 基础查询

python

async def get_user_by_id(user_id: int):
async with async_session() as session:
# select() 是 2.0 风格的核心查询入口
stmt = select(User).where(User.id == user_id)
result = await session.execute(stmt)
return result.scalar_one_or_none()

3.2 条件组合与排序

python

from sqlalchemy import select, and_, or_

async def search_users(name: str = None, email: str = None):
async with async_session() as session:
conditions = []
if name:
conditions.append(User.name.ilike(f"%{name}%"))
if email:
conditions.append(User.email.ilike(f"%{email}%"))

stmt = select(User).where(and_(*conditions)).order_by(User.name.asc())
result = await session.execute(stmt)
return result.scalars().all()

3.3 聚合查询

使用 func 进行分组统计：

python

from sqlalchemy import func

async def get_post_count_by_user():
async with async_session() as session:
stmt = (
select(User.name, func.count(Post.id).label("post_count"))
.join(Post)
.group_by(User.id, User.name)
.order_by(func.count(Post.id).desc())
)
result = await session.execute(stmt)
return result.all() # 返回 [(name, count), …]

3.4 分页

python

async def get_posts_paginated(page: int, page_size: int = 10):
async with async_session() as session:
offset = (page – 1) * page_size
stmt = select(Post).offset(offset).limit(page_size).order_by(Post.id.desc())
result = await session.execute(stmt)
return result.scalars().all()

四、关系加载：告别 N+1 查询噩梦

关系加载是 ORM 性能的关键。默认情况下，relationship 使用延迟加载（lazy="select"）——访问关联属性时才发 SQL。这在循环中会引发经典的 N+1 查询问题 。

4.1 三种加载策略对比

4.2 使用 Select IN 加载（最推荐）

python

from sqlalchemy.orm import selectinload

async def get_users_with_posts():
async with async_session() as session:
# 一次查询用户，第二次用 IN 批量加载所有关联帖子
stmt = select(User).options(selectinload(User.posts))
result = await session.execute(stmt)
return result.scalars().all()

selectinload 是 2.0 时代最推荐的预加载方式——它总是发出第二条独立的 SELECT 语句，不会像 joinedload 那样改变主查询的行数 。

4.3 链式加载深层关系

python

stmt = select(Post).options(
selectinload(Post.author), # 加载作者
selectinload(Post.comments).selectinload(Comment.replies) # 加载评论及其回复
)

链式调用可以精确控制每一层关系的加载策略，避免加载不需要的数据 。

4.4 防止意外的延迟加载：raiseload

如果你希望彻底杜绝 N+1 问题，可以使用 raiseload——一旦触发延迟加载就抛出异常 ：

python

from sqlalchemy.orm import raiseload

stmt = select(User).options(raiseload(User.posts))
# 访问 user.posts 会抛异常，强制你提前用 selectinload

五、事务管理：提交、回滚与保存点

5.1 基础事务

Session 默认使用“自动开始”模式——执行第一条 SQL 时自动开启事务。推荐用 async with session.begin() 显式划定事务边界 ：

python

async def create_user_with_posts(user_data, posts_data):
async with async_session() as session:
async with session.begin(): # 事务开始，结束时自动提交
user = User(**user_data)
session.add(user)
# flush 后 user.id 才可用，但不会提交
await session.flush()
for post_data in posts_data:
session.add(Post(author_id=user.id, **post_data))
# 退出 begin 块时自动 commit，异常则 rollback

5.2 SAVEPOINT：部分回滚

在复杂业务中，你可能想“部分回滚”而不用放弃整个事务。begin_nested() 创建保存点 ：

python

async def batch_create_users(records):
async with async_session() as session:
async with session.begin():
for record in records:
try:
async with session.begin_nested(): # 保存点
user = User(**record)
session.add(user)
except IntegrityError: # 比如邮箱重复
print(f"跳过重复记录: {record['email']}")
# 外层事务仍然可以提交

5.3 事务的两种风格

• 逐次提交（commit-as-you-go）：显式调用 commit() 

• 一次开始（begin-once）：用 begin() 块包裹，推荐更安全

六、异步编程：AsyncSession 的正确姿势

6.1 基本结构

SQLAlchemy 2.0 的异步支持基于 asyncpg 或 aiosqlite 等异步驱动。核心是 AsyncSession ：

python

async def main():
async with async_session() as session:
# 所有数据库操作都需要 await
stmt = select(User).where(User.name == "Alice")
result = await session.execute(stmt)
user = result.scalar_one()

# 修改后提交
user.email = "alice@new.com"
await session.commit()

6.2 关键注意事项：避免隐式 IO

异步编程中最容易踩的坑是：访问未预加载的关系属性时，触发延迟加载——但延迟加载是同步的！

python

# ❌ 错误：访问 user.posts 会触发同步延迟加载，导致事件循环阻塞
user = await session.get(User, 1)
print(user.posts) # 这里会发同步 SQL！

# ✅ 正确：提前用 selectinload 预加载
stmt = select(User).where(User.id == 1).options(selectinload(User.posts))
user = (await session.execute(stmt)).scalar_one()
print(user.posts) # 已加载，安全

6.3 使用 AsyncAttrs 延迟加载

如果确实需要在异步上下文中触发延迟加载，可以继承 AsyncAttrs 并使用 awaitable_attrs ：

python

class Base(AsyncAttrs, DeclarativeBase):
pass

# 使用时
user = await session.get(User, 1)
posts = await user.awaitable_attrs.posts # 异步延迟加载

6.4 run_sync：同步代码的桥接

有些 ORM 操作（如 metadata.create_all()）没有异步版本，可以用 run_sync 桥接 ：

python

async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)

七、进阶技巧

7.1 Write-Only 关系（内存友好）

对于超大集合，write-only 模式不会在内存中保留全部关联对象，只有显式查询时才会加载 ：

python

from sqlalchemy.orm import WriteOnlyMapped

class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
posts: WriteOnlyMapped["Post"] = relationship()

async def get_user_posts(user_id: int):
async with async_session() as session:
user = await session.get(User, user_id)
stmt = user.posts.select().where(Post.published == True)
result = await session.execute(stmt)
return result.scalars().all()

7.2 批量操作

python

# 批量插入
async def bulk_create_users(users_data):
async with async_session() as session:
async with session.begin():
session.add_all([User(**data) for data in users_data])

八、总结

本文覆盖了 SQLAlchemy ORM 的四大支柱：

SQLAlchemy 2.0 不再是一个“功能强大但笨重”的框架。它的异步支持已经非常成熟，类型注解也让代码更加清晰可靠。掌握以上内容，你基本可以应对 80% 以上的日常数据库开发场景。