如何在SQLAlchemy中对加密列使用SUM函数？

我使用sqlalchemy_utils的EncryptedType加密了数据库特定列的数据，ORM结构如下，其中value列是加密的整数类型：

from sqlalchemy_utils import EncryptedType
from sqlalchemy_utils.types.encrypted.encrypted_type import AesEngine

class Products(db.Model):
    __tablename__ = 'products'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(400))
    value = db.Column(EncryptedType(db.Integer, secret_key, AesEngine,'pkcs5')) 

数据库中value列存储为加密后的bytea格式：

id |  name   |                       value                        
----+---------+----------------------------------------------------
  1 | Macbook | \x6977764a59556346536e6b674d7a6439312f714c70413d3d
  2 | IPhone  | \x6a6b51757a48554739666756566863324662323962413d3d
  3 | IPad    | \x416d54504b787873462f724d347144617034523639673d3d

通过ORM查询单条数据时，value会自动解密：

product_query = Products.query.order_by(Products.id.asc()).all()
for product in product_query:
    print(product.id, ' ', product.name, ' ', product.value)

输出结果：

1   Macbook   222222
2   IPhone   40000
3   IPad   60000

但执行求和操作时：

db.session.query(func.sum(Products.value)).all()

出现报错：

sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedFunction) function sum(bytea) does not exist
LINE 1: SELECT sum(products.value) AS sum_1 

原因是数据库中value为bytea类型，无法直接进行求和运算，需要找到实现加密列数值求和的方法。

针对该问题，有两种可行的处理方式：

方式一：Python层面求和 #

先通过ORM查询所有加密列数据（此时会自动解密），再在Python内存中计算总和：

products = Products.query.with_entities(Products.value).all()
total = sum(p.value for p in products)

这种方式简单直接，适合数据量较小的场景；缺点是需要将所有数据加载到内存中，数据量大时会占用较多资源。

方式二：数据库端解密后求和 #

如果数据量较大，不想全量加载到内存，可以自定义支持数据库端解密的类型，让SQLAlchemy生成解密SQL后再对数值求和。

以PostgreSQL为例，需先确保数据库安装pgcrypto扩展，然后自定义类型：

from sqlalchemy_utils.types.encrypted.encrypted_type import EncryptedType
from sqlalchemy_utils.types.encrypted.encrypted_type import AesEngine
from sqlalchemy import func

class PostgresEncryptedType(EncryptedType):
    def expression(self, column):
        # 生成PostgreSQL解密SQL，指定加密算法、模式和填充方式
        decrypt_func = func.pgp_sym_decrypt(column, self.key, f'cipher-algo=aes256,mode=cfb,padding={self.padding}')
        # 将解密结果转换为对应的数据类型（此处为Integer）
        return func.cast(decrypt_func, self.type)

修改ORM模型的value列定义：

value = db.Column(PostgresEncryptedType(db.Integer, secret_key, AesEngine,'pkcs5'))

之后即可直接使用func.sum进行求和：

total = db.session.query(func.sum(Products.value)).scalar()

注意：该方式要求数据库支持对应解密函数，且密钥会带入SQL语句中，需注意SQL注入风险和密钥安全性。

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