sqlalchemy.orm中validates对两个字段进行联合校验

版本

sqlalchemy==1.4.37

需求说明

有个场景，需要在orm中对两个字段进行联合校验，当 col1 ='xxx’时，对 col2的长度进行检查，超过限制（500）时，进行截断。

网上找了很久，没找到类似的实现，自己摸索出来了一套方法；

解决

在 validates 装饰器中，它是在设置字段值之前被调用的，validates 包装的函数校验完成后通过return赋值给字段

validates 的执行顺序 看起来是和 字段 传入ORM模型的顺序 一样。
如 下面案例 中的 model_instance 中，如果 col1 在 col2 之前，就会先校验和赋值 col1 ，反之，则先校验和赋值 col12

方案1

保证 model_instance 中 字段 col1 在 col2 之前，这样会先校验和赋值 col1 ，在校验 col2 时，self.col1 就不会为空，能正常进行校验。

如果先赋值col2，在 validate_col2 中，会self.col1会为None，导致校验失败

from sqlalchemy.orm import validates
from sqlalchemy import Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class MyModel(Base):
    __tablename__ = 'my_model'
    id = Column(Integer, primary_key=True)
    col1 = Column(String(50))
    col2 = Column(String(500))  # 假设col2的最大长度是500个字符

    @validates('col2')
    def validate_col2(self, key, value):
        # 检查col1的值是否为'xxx'
        if self.col1 == 'xxx':
            # 如果col1是'xxx'，则校验col2的长度
            if len(value) > 500:
                value = value[:500]
        # 如果col1不是'xxx'，可以选择不做任何操作或者添加其他逻辑
        return value 

# 示例使用1
# 先赋值 col1 再赋值 col2
model_instance = MyModel()
model_instance.col1 = 'xxx'  # 假设这是触发条件的值
model_instance.col2 = 'a' * 501  # 这将触发长度校验

# 示例使用2
datas = {'col1'= 'xxx', 'col2': 'a' * 501 }
# 先pop删除，再添加，就不管前面datas是怎么来的，可以保证 datas 中 col2会比col1后遍历到
_col2_v = datas.pop('col2')
datas['col2'] = _col2_v  
model_instance = MyModel(**datas )


try:
    # 假设这是保存模型到数据库的代码
    # session.add(model_instance)
    # session.commit()
    pass
except ValueError as e:
    print(e)

方案2

无法保证 model_instance 中 字段 col1 在 col2 之前的顺序，采用 临时变量 __col1，存储 col1 的值，并对 col2 进行二次校验赋值

在 validate_col1 函数中，校验 col1，先把 value 值（就是没校验前的col1的值）赋给 self.__col1，然后再调用 validate_col1_and_col2 进行联合校验，最后通过 return把value赋值给 self.col1

在整个过程中，validate_col1_and_col2 会被调用3次

• 校验 col2 时，validate_col2 会调用一次
• 校验 col1 时，self.col2 = self.validate_col1_and_col2(key='col2', value=self.col2) 这一行会调用两次:
  1. 一次是 self.validate_col1_and_col2 执行；
  2. 另一次是 1 执行完后对 self.col2 赋值，会调用一次 validate_col2 ，进而再调用一次

from sqlalchemy.orm import validates
from sqlalchemy import Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base



def getStrLenAndTruncate(ss: str, max_length=500):
    """
    获取字符串长度，超过部分截断
    :param ss:
    :param max_length:
    :return:
    """
    slen = len(ss.encode('utf-8'))

    # 如果编码后的字符串长度小于或等于500字节，则不需要截断
    if slen <= max_length:
        return ss

    # 截断到500字节的长度，注意这里直接截断可能会导致字符不完整
    # 因此需要找到一个合适的截断点，确保截断后的字符串是完整的utf-8字符
    truncated_encoded = b''
    current_length = 0
    for char in ss:
        char_encoded = char.encode('utf-8')
        if current_length + len(char_encoded) <= max_length:
            truncated_encoded += char_encoded
            current_length += len(char_encoded)
        else:
            break

    truncated_str = truncated_encoded.decode('utf-8', errors='ignore')
    print(f'原字符串编码后长度为{slen}, 超过限制{max_length}, 需进行截断\n原字符串={ss}, \n截断后字符串:{truncated_str}')
    return truncated_str


Base = declarative_base()


class MyModel(Base):
    __tablename__ = 'my_model'
    __col1 = ''
    id = Column(Integer, primary_key=True)
    col1 = Column(String(50))
    col2 = Column(String(500))  # 假设col2的最大长度是500个字符
	

    @validates('col1')
    def validate_col1(self, key, value):
        self.__col1 = value
        self.col2 = self.validate_col1_and_col2(key='col2', value=self.col2)
        return value
        

    @validates('col2')
    def validate_col2(self, key, value):
        value = validate_col1_and_col2(key, value)
        return value 
    
    def validate_col1_and_col2(self, key, value):
        # 检查col1的值是否为'xxx'
    	if self.__col1== 'xxx':
    	    if not value:
                value = ''
            elif len(value) * 3 <= 500:
                # 存储到 oracle，中文占3个字符
                pass
            else:
                print('需检查 col2 长度')
                value = getStrLenAndTruncate(value)

        return value
 
# 示例使用
datas = {'col1'= 'xxx', 'col2': 'a' * 501 }
model_instance = MyModel(**datas )


try:
    # 假设这是保存模型到数据库的代码
    # session.add(model_instance)
    # session.commit()
    pass
except ValueError as e:
    print(e)

说明

为啥不省略下面这个 validate_col2 这个校验代码：

    @validates('col2')
    def validate_col2(self, key, value):
        value = validate_col1_and_col2(key, value)
        return value 

因为 这个方案中，col1 、col2进入 orm模型的顺序不一定，如果省略了validate_col2 ，当col1比col2先进入模型，那么在 validate_col1 调用 self.validate_col1_and_col2(key='col2', value=self.col2) 时，self.col2其实等于None，此时对col2校验是没有意义的。等到 col2 进入 orm模型，又缺少对它进行校验的函数。

注意：

不能在 某个字段的校验函数中对其进行赋值操作，不然会陷入递归循环，因为赋值操作会调用校验函数；

如下面的调用会陷入递归死循环，因为 self.col1 = value 这行代码对 self.col1 进行了赋值，会自动再次调用validate_col1校验函数，就会在这一行陷入递归死循环而报错。

class MyModel(Base):
    @validates('col1')
    def validate_col1(self, key, value):
        self.__col1 = value
        self.col1 = value
        self.col2 = self.validate_col1_and_col2(key='col2', value=self.col2)
        return value
        

其他方案

不走orm模型，直接写校验代码和原生sql处理。

最后

这就是我尝试出来的 在 sqlalchemy.orm中validates对两个字段进行联合校验的方法，总感觉不太完美，不知道有没有大佬知道更好的方案，欢迎分享