
SQLAlchemy 默认会对所有数据库操作(包括 SELECT)隐式启动事务,即使仅执行读取查询;它不提供原生 readOnly=true 机制,但可通过事件钩子和会话元数据模拟真正只读会话,防止意外写入与提交。
sqlalchemy 默认会对所有数据库操作(包括 select)隐式启动事务,即使仅执行读取查询;它不提供原生 `readonly=true` 机制,但可通过事件钩子和会话元数据模拟真正只读会话,防止意外写入与提交。
在 SQLAlchemy 中,事务并非按“操作类型”(读/写)动态启用,而是按“会话活跃状态”统一管理。官方文档中提到的“virtual transaction”(虚拟事务)指的是:当 Session 首次执行任何数据库语句(如 session.execute()、session.query() 或 session.add())时,SQLAlchemy 会自动开启一个事务(BEGIN)——无论该语句是 SELECT 还是 INSERT。这意味着:
✅ session.execute(select(...)) 一定会触发事务启动(底层发送 BEGIN);
✅ 即使后续仅做查询,该事务仍处于活跃状态,直到显式调用 commit()、rollback(),或 Session 被关闭;
❌ 不存在“无事务的纯读取模式”——SQLAlchemy 不支持跳过事务启动的“只读轻量级连接”。
这与 Spring Data JPA 的 @Transactional(readOnly = true) 有本质区别:后者虽标记为只读,但仍会开启事务(通常以 READ COMMITTED 隔离级别),但可触发数据库优化(如 Oracle 的 SET TRANSACTION READ ONLY);而 SQLAlchemy 不提供等效的声明式只读事务配置,也不向数据库发送只读事务指令(除非手动通过 execution_options(isolation_level=...) 设置)。
不过,我们可以通过组合使用 Session.info 元数据与 @event.listens_for(Session, "before_flush") 实现逻辑层面的只读会话保护,如下所示:
from sqlalchemy import create_engine, Column, Integer, String, select
from sqlalchemy.orm import Session, sessionmaker, declarative_base
from sqlalchemy.event import listen
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(50))
def set_read_only(session: Session, read_only: bool = True):
session.info["read_only"] = read_only
def is_read_only(session: Session) -> bool:
return session.info.get("read_only", False)
# 阻止 flush 和 commit
@listen.on("before_flush", Session)
def prevent_flush(session, flush_context, instances):
if is_read_only(session):
raise RuntimeError("Flush denied: session is read-only")
@listen.on("do_orm_execute", Session)
def prevent_commit_or_rollback(context):
if is_read_only(context.session) and context.is_commit or context.is_rollback:
raise RuntimeError("Commit/rollback denied: session is read-only")
✅ 关键增强点:相比原始示例,此处补充了
do_orm_execute事件监听,拦截session.commit()和session.rollback()调用,确保只读会话无法被意外提交,提升安全性。
使用时只需在获取 Session 后标记即可:
SessionLocal = sessionmaker(bind=engine) session = SessionLocal() set_read_only(session) # 启用只读防护 # ✅ 安全:查询正常执行 users = session.execute(select(User)).scalars().all() # ❌ 报错:add() 触发 flush → before_flush 拦截 # session.add(User(name="Bob")) # ❌ 报错:commit 被显式拦截 # session.commit()
⚠️ 注意事项:
- 此方案是应用层防护,不改变底层 SQL 行为(
BEGIN仍会发生),但能有效避免脏写与误提交; - 若需极致性能(如高并发只读报表),建议配合数据库连接池配置专用只读副本(如 PostgreSQL 的
pgbouncer+replica连接字符串),而非依赖 ORM 层控制; - 在 FastAPI 中,推荐通过依赖注入封装只读 Session 工厂:
def get_readonly_session(): session = SessionLocal() set_read_only(session) try: yield session finally: session.close()
总结:SQLAlchemy 总是为首次数据库访问启动事务,没有“零事务只读”模式;但通过事件驱动 + 会话元数据,可构建健壮、易复用的逻辑只读会话,兼顾安全性与开发体验——这是贴近生产实践的务实解法。










