SQLAlchemy自定义PGPString类型如何让WHERE子句解密列而非加密参数

你当前的自定义类型逻辑里，bind_expression负责加密写入参数、column_expression负责解密查询列，但SQLAlchemy生成比较条件时会默认对齐两侧数据类型：当你写User.username == '明文'时，SQLAlchemy会自动将右侧的明文参数通过bind_expression加密，再和左侧未解密的原始密位列比对，pgp_sym_encrypt自带随机盐，相同明文每次加密结果不同，自然匹配不到结果。

通过自定义比较器comparator_factory覆盖默认的比较逻辑，让比较操作优先使用解密后的列和明文参数直接比对，不再对右值参数加密。完整修改后的代码如下：

from sqlalchemy import TypeDecorator, String, func
from sqlalchemy.sql.elements import BindParameter

# 替换为你实际使用的加密密钥，建议从配置文件读取不要硬编码
PGP_SECRET_KEY = "your_pgp_secret_key"

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

    def bind_expression(self, bindvalue):
        # 写入逻辑保持不变，依然对参数加密
        return func.pgp_sym_encrypt(bindvalue, PGP_SECRET_KEY)

    def column_expression(self, col):
        # 列查询逻辑保持不变，默认返回解密后的明文
        return func.pgp_sym_decrypt(col, PGP_SECRET_KEY)
    
    class comparator_factory(TypeDecorator.Comparator):
        # 自定义比较逻辑，覆盖等于操作
        def __eq__(self, other):
            if isinstance(other, (str, BindParameter)):
                # 直接拿解密后的列和明文参数比对，不对参数加密
                return self.expr == other
            return super().__eq__(other)
        
        # 按需覆盖不等于逻辑
        def __ne__(self, other):
            if isinstance(other, (str, BindParameter)):
                return self.expr != other
            return super().__ne__(other)
        
        # 如果需要模糊查询，可继续覆盖like、ilike等方法，逻辑和上述一致

修改后执行查询：

session.query(User).filter(User.username == "test_user").first()

生成的WHERE子句会自动变成你需要的形式：
WHERE pgp_sym_decrypt(user.username, 'your_pgp_secret_key') = 'test_user'
原有加密写入的功能不受影响。

该方案是对全列解密后再做值比对，数据量较大时会无法走索引、查询性能较差。如果需要优化性能，可以额外存储明文字段的加盐哈希值做索引，查询时先匹配哈希值再做解密校验，避免全表扫描。

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