如何为基于TypeDecorator的自定义ORM类型所有列统一添加set事件监听器

我基于SQLAlchemy的TypeDecorator实现了自定义数据类型PathType，可将字符串转换为pathlib.Path对象。需要为所有使用该类型的列监听set事件，但当前实现耦合了Base类，希望直接从绑定PathType的Column对象出发实现，做到关注点分离。

当前实现 #

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.types import TypeDecorator, String
from pathlib import Path
from sqlalchemy import event

class PathType(TypeDecorator):
    impl = String
    cache_ok = True

    def process_bind_param(self, value: Path, dialect):
        if value is not None:
            value = str(value.resolve())
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = Path(value)
        return value

class Base(DeclarativeBase):
    pass

class Model(Base):
    __tablename__ = "model"
    path: Mapped[Path] = mapped_column(PathType()) 

def resolve_values_assigned_to_path_type():
    @event.listens_for(Base, 'attribute_instrument')
    def configure_listener(table_cls, attr, col_inst):
        if not hasattr(col_inst.property, 'columns'):
            return
        if isinstance(col_inst.type, PathType):
            @event.listens_for(col_inst, "set", retval=True)
            def receive_set(target, value: Path, oldvalue, initiator):
                return value.resolve()

model = Model()
resolve_values_assigned_to_path_type()

期望目标 #

解除对Base类的依赖，直接从绑定PathType的Column对象出发实现set事件监听，实现关注点分离。

尝试方案 #

曾尝试监听before_parent_attach事件，虽能获取对应Table.Column，但无法关联到ORM属性，无法绑定set事件：

@event.listens_for(PathType, "before_parent_attach")
def evt(typ, col):
    @event.listens_for(col, "after_parent_attach")
    def evt2(col, table):
        @event.listens_for(col, "set", retval=True)
        def receive_set(target, value: Path, oldvalue, initiator):
            return value.resolve()

你之前的尝试无法生效，是因为set事件是针对**ORM属性（ColumnProperty）**而非数据库表的Column对象的。要实现不依赖Base类的全局监听，可以利用SQLAlchemy的mapper_configured事件，在Mapper完成配置后，遍历所有属性并为PathType类型的属性绑定set事件：

from sqlalchemy import event
from sqlalchemy.orm import Mapper
from pathlib import Path

# 保留原PathType和Base、Model定义不变

@event.listens_for(Mapper, "mapper_configured")
def setup_path_type_listeners(mapper, class_):
    # 遍历当前Mapper的所有属性
    for attr in mapper.attrs:
        # 检查属性是否关联了数据库列
        if hasattr(attr, "columns") and attr.columns:
            column = attr.columns[0]
            # 判断列类型是否为PathType
            if isinstance(column.type, PathType):
                # 为该ORM属性绑定set事件
                @event.listens_for(attr, "set", retval=True)
                def handle_path_set(target, value: Path, oldvalue, initiator):
                    if value is not None:
                        return value.resolve()
                    return value

# 实例化模型时自动触发监听配置
model = Model()

原理说明 #

1. mapper_configured事件会在每个ORM类的Mapper完成配置后触发，无需绑定具体Base类，覆盖所有使用DeclarativeBase的子类。
2. 遍历Mapper的所有属性，筛选出类型为PathType的列对应的ORM属性。
3. 为符合条件的ORM属性绑定set事件，在属性被赋值时自动解析Path对象。

这样既实现了关注点分离，无需依赖具体的Base类，又能全局为所有PathType列添加set事件监听。

内容的提问来源于stack exchange，提问作者aktabit