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_baseBase = 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

实例化orm模型后，再调用一遍 validate_col2 ，校验并赋值给col2

from sqlalchemy.orm import validates
from sqlalchemy import Column, String, Integer
from sqlalchemy.ext.declarative import declarative_baseBase = 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 }model_instance = MyModel(**datas )
# 上面实例化会自动调用所有的 `validates` 函数，下面再调用一遍`validate_col2`, 
# 并且要用 `model_instance.col2` 接收返回值，不然 `model_instance.col2` 的值不会改变。
model_instance.col2 = model_instance.validate_col2('col2', datas['col2'])try:# 假设这是保存模型到数据库的代码# session.add(model_instance)# session.commit()pass
except ValueError as e:print(e)

方案3

无法保证 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_basedef 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 = 0for char in ss:char_encoded = char.encode('utf-8')if current_length + len(char_encoded) <= max_length:truncated_encoded += char_encodedcurrent_length += len(char_encoded)else:breaktruncated_str = truncated_encoded.decode('utf-8', errors='ignore')print(f'原字符串编码后长度为{slen}, 超过限制{max_length}, 需进行截断\n原字符串={ss}, \n截断后字符串:{truncated_str}')return truncated_strBase = 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 = valueself.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个字符passelse: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 = valueself.col1 = valueself.col2 = self.validate_col1_and_col2(key='col2', value=self.col2)return value

其他方案

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

最后

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