本文介绍使用 SQLAlchemy 2.0+ 的 do_orm_execute 事件钩子，在查询执行前动态注入业务逻辑（如根据 show_value_a 自动掩码 value_a），实现声明式、零侵入、Alembic 友好的字段级权限/展示控制。
本文介绍使用 sqlalchemy 2.0+ 的 `do_orm_execute` 事件钩子，在查询执行前动态注入业务逻辑（如根据 `show_value_a` 自动掩码 `value_a`，实现声明式、零侵入、alembic 友好的字段级权限/展示控制。
在构建数据访问层时，常需将业务规则（如“仅当 show_value_a 为 True 时才暴露 value_a”）与底层 ORM 模型解耦——既不能污染表定义类（避免与 Alembic 迁移职责混杂），又需确保所有查询（无论是否含 JOIN、FILTER 或 ORDER BY）统一生效。SQLAlchemy 1.4（启用 future=True）及 2.x 提供的 do_orm_execute 事件正是为此场景设计：它允许你在 ORM 查询真正编译执行前，安全地检查、修改 SQL 表达式树。
核心思路是：监听 Session 级别的 do_orm_execute 事件 → 判断当前 SELECT 是否涉及需受控的实体（如 MyTable）→ 动态重写目标列（如 valueA）为带条件表达式的 CASE WHEN showValueA THEN valueA ELSE NULL END → 替换原语句并继续执行。
以下是一个生产就绪的实现示例：
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.orm import Mapped, mapped_column
class Base(orm.DeclarativeBase):
pass
class MyTable(Base):
__tablename__ = "mytable"
id: Mapped[int] = mapped_column(primary_key=True)
valueA: Mapped[str] = mapped_column("value_a")
showValueA: Mapped[bool] = mapped_column("show_value_a")
# 配置 Session 工厂，并预设需监控的实体列表
engine = sa.create_engine("sqlite:///example.db", echo=True)
Base.metadata.create_all(engine)
# 通过 session.info 传递配置，避免全局状态
Session = orm.sessionmaker(
engine,
info={"check_entities": {MyTable}} # 支持多实体，如 {User, Order, MyTable}
)
@sa.event.listens_for(Session, "do_orm_execute")
def _do_orm_execute(orm_execute_state):
if not orm_execute_state.is_select:
return # 仅处理 SELECT
statement = orm_execute_state.statement
# 安全检测：仅当查询明确包含 MyTable 实体时才介入
if not any(
desc["entity"] is MyTable
for desc in statement.column_descriptions
if desc.get("entity")
):
return
# 构建条件表达式：CASE WHEN showValueA THEN valueA ELSE NULL
masked_value_a = sa.case(
(MyTable.showValueA, MyTable.valueA),
else_=None
).label("value_a")
# 替换原语句中的 value_a 列（保留其他列不变）
new_columns = []
for col in statement.inner_columns:
if hasattr(col, "name") and col.name == "value_a":
new_columns.append(masked_value_a)
else:
new_columns.append(col)
# 重建 SELECT 语句，复用原语句的 FROM/WHERE/ORDER BY/JUMP 等结构
# ⚠️ 注意：此处为简化示例；实际项目中建议使用 statement.transform() 或深度克隆
new_statement = sa.select(*new_columns).select_from(statement.froms[0])
# 继承原查询的 WHERE / ORDER BY / LIMIT 等子句（关键！）
if statement.whereclause is not None:
new_statement = new_statement.where(statement.whereclause)
if statement.order_by_clauses:
new_statement = new_statement.order_by(*statement.order_by_clauses)
if statement.limit_clause is not None:
new_statement = new_statement.limit(statement.limit_clause)
orm_execute_state.statement = new_statement✅ 优势与保障：
-
职责分离：
MyTable仅负责映射，业务逻辑由事件处理器集中管理； - Alembic 安全：事件注册在运行时，不修改模型定义，迁移脚本完全不受影响；
-
查询兼容：支持任意复杂查询（JOIN、子查询、聚合），只要最终投影包含
MyTable.valueA； -
可扩展：通过
session.info['check_entities']动态控制作用域，不同 Session 可启用不同规则。
⚠️ 注意事项：
- 该方案要求使用 SQLAlchemy 2.0 风格（或 1.4 启用
future=True）； - 若查询使用
load_only()或defer()，需额外处理列投影逻辑； - 对性能敏感场景，建议对
do_orm_execute添加轻量缓存或短路判断（如检查statement.column_descriptions是否为空）； - 不适用于纯 Core 查询（
connection.execute()），仅作用于 ORM 查询（session.scalars()/session.query()）。
最后验证效果：
with Session() as s:
# 原始写法，自动生效
results = s.scalars(sa.select(MyTable.valueA)).all()
print(results) # 输出: ['A', None, 'C']
# 复杂查询同样适用
complex_q = (
sa.select(MyTable.valueA, MyTable.showValueA)
.where(MyTable.id > 1)
.order_by(MyTable.id)
)
for val, show in s.execute(complex_q).all():
print(val, show) # val 已自动掩码此模式将“数据可见性”提升为可插拔的横切关注点，显著降低重复逻辑和漏配风险，是构建健壮业务数据访问层的推荐实践。
