Python2 NDB查询Cloud NDB创建的实体时间歇性断言错误求助

问题根源 #

Cloud NDB与旧版Python2 NDB对**空重复属性（RepeatedProperty）**的序列化格式存在差异：Cloud NDB会将空重复字段存储为空列表[]，而旧NDB的_BaseValue初始化逻辑中存在断言，禁止传入列表类型值，导致get()操作时触发AssertionError: []。

解决方案 #

1. 猴子补丁修复旧NDB的断言逻辑 #

由于无法直接修改App Engine官方的旧NDB库代码，可以在服务启动时通过猴子补丁覆盖_BaseValue的初始化方法，兼容空列表：

from google.appengine.ext.ndb import model

original_base_value_init = model._BaseValue.__init__

def patched_base_value_init(self, b_val):
    # 处理Cloud NDB传入的空列表
    if isinstance(b_val, list):
        # 判断当前属性是否为重复类型
        if hasattr(self, '_base_type') and isinstance(self._base_type, model.RepeatedProperty):
            # 重复属性允许列表值，直接保留
            pass
        else:
            # 非重复属性，空列表转为None，非空列表取第一个元素
            b_val = b_val[0] if b_val else None
    # 调用原初始化方法
    original_base_value_init(self, b_val)

# 替换原方法
model._BaseValue.__init__ = patched_base_value_init

将这段代码放在服务入口文件（如main.py）的最顶部，确保在任何NDB操作前加载。

2. 修改Cloud NDB存储逻辑，兼容旧NDB #

在Python3的Cloud NDB代码中，保存实体前清理空的重复字段，避免存储空列表：

from google.cloud import ndb

class YourModel(ndb.Model):
    repeated_field = ndb.StringProperty(repeated=True)
    # 其他字段...

    def put(self, *args, **kwargs):
        # 遍历所有重复属性，清空则删除该字段
        for prop_name, prop in self._properties.items():
            if isinstance(prop, ndb.RepeatedProperty):
                val = getattr(self, prop_name, None)
                if not val:
                    delattr(self, prop_name)
        super().put(*args, **kwargs)

这样Cloud NDB创建的实体不会存储空重复字段，旧NDB的get()操作就不会收到触发断言的空列表。

3. 封装稳定的替代get方法（过渡方案） #

如果暂时无法修改库或存储逻辑，可封装自己的safe_get方法，异常时自动 fallback 到查询：

from google.appengine.ext.ndb import Key

def safe_get(key):
    try:
        return key.get()
    except AssertionError as e:
        if str(e) == "[]":
            # 触发目标断言错误时，用查询获取实体
            return key.query(Key == key).get()
        # 其他断言错误正常抛出
        raise

或者使用Datastore低级API直接获取并转换实体：

from google.appengine.api import datastore
from google.appengine.ext.ndb import model

def safe_get(key):
    # 用低级API获取原始实体
    raw_entity = datastore.Get(key.to_old_key())
    if not raw_entity:
        return None
    # 转换为NDB模型实例
    return model.Model._from_entity(raw_entity, key=key)

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