如何在SQLAlchemy中实现支持写同步的带过滤关联属性

实现方案 #

你可以通过自定义一个过滤代理集合类的方式实现需求，该方案会完全复用原有items关联的逻辑，天然支持双向同步，且兼容所有列表操作API：

第一步：实现通用过滤代理类 #

from sqlalchemy.ext.orderinglist import ordering_list
from collections.abc import MutableSequence

class FilteredContentTypeProxy(MutableSequence):
    def __init__(self, origin_list, content_type):
        self.origin_list = origin_list
        self.content_type = content_type

    # 内部过滤出符合类型的元素
    def _filtered_items(self):
        return [item for item in self.origin_list if item.type == self.content_type]

    # 实现列表读取相关的魔法方法
    def __getitem__(self, index):
        return self._filtered_items()[index]
    def __len__(self):
        return len(self._filtered_items())
    def __iter__(self):
        return iter(self._filtered_items())
    def __contains__(self, item):
        return item in self._filtered_items()

    # 实现列表写入相关的魔法方法，所有操作都代理到原集合
    def __setitem__(self, index, value):
        value.type = self.content_type
        filtered = self._filtered_items()
        origin_idx = self.origin_list.index(filtered[index])
        self.origin_list[origin_idx] = value
    def __delitem__(self, index):
        filtered = self._filtered_items()
        origin_idx = self.origin_list.index(filtered[index])
        del self.origin_list[origin_idx]
    def insert(self, index, value):
        value.type = self.content_type
        filtered = self._filtered_items()
        if index >= len(filtered):
            self.origin_list.append(value)
        else:
            origin_idx = self.origin_list.index(filtered[index])
            self.origin_list.insert(origin_idx, value)
    def append(self, value):
        value.type = self.content_type
        self.origin_list.append(value)
    def remove(self, value):
        if value in self._filtered_items():
            self.origin_list.remove(value)

第二步：修改Collection类的属性定义 #

class Collection(db.Model):
    id = db.Column(db.Integer(), primary_key=True)
    items = db.relationship(
        "Content",
        backref="collection",
        collection_class=ordering_list("weight")
    )

    # 定义poems属性，返回过滤代理实例
    @property
    def poems(self):
        return FilteredContentTypeProxy(self.items, content_type='poem')

方案说明 #

• 所有读写操作都直接代理到原items关联集合，不存在多关联数据不同步的问题
• 写入时会自动为元素设置对应的type值，无需手动传入，避免逻辑错误
• 完全复用items的ordering_list排序逻辑，weight字段的排序行为和原集合完全一致
• 支持所有原生列表操作：迭代、索引访问、切片、append、remove、len计算等
• 扩展成本极低，如果需要新增其他类型的过滤属性，仅需新增一行property定义即可，例如：

@property
def novels(self):
    return FilteredContentTypeProxy(self.items, content_type='novel')

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