在googleappengine（Python）中将数据从产品数据存储传输到本地开发环境数据存储

Question

我需要找到一个真正的解决方案来从产品数据存储中下载数据并将其加载到本地开发环境中。在

具体问题：

我需要在本地开发服务器上用产品服务器数据存储上的真实数据（不是实时数据）测试我的应用程序。文件和其他资源提供了三种选择：

1. 使用appfg.py从产品服务器下载数据，然后将其加载到本地开发环境中。当我使用此方法时，由于Oauth问题，我收到“bad request”错误。此外，此方法将被弃用。官方文件建议使用第二种方法：
2. 使用gcloud via managed export and the import。这个方法的epic文档解释了我们如何在控制台（在https://console.cloud.google.com/）上备份所有数据。我试过这种方法。备份数据正在云中的存储中生成。我下载了。它是LevelDB格式。我需要把它加载到本地开发服务器。官方对此没有任何解释。第一种方法的加载方法与LevelDB格式不兼容。我找不到一个官方的方法来解决这个问题。有一个StackOverflow entry但它对我不起作用，因为它只是把所有的实体都当作dict来处理。对话中的'dic'对象到'ndb'实体就成了一个棘手的问题。在
3. 我对前两种方法已经失去了希望，然后我决定使用Cloud Datastore Emulator (beta)，它在本地开发环境中提供模拟真实数据。它仍然是测试版，有几个问题。当我运行该命令时，我遇到了问题DATASTORE\u EMULATOR_HOST。在

Answer 1

听起来你应该使用远程沙盒

即使您使其正常工作，localhost数据存储的行为仍然与实际数据存储不同。在

如果您想真正模拟您的生产环境，那么我建议您将appengine项目的克隆设置为远程沙箱。您可以将应用程序部署到新的gae项目id appcfg.py update . -A sandbox-id，然后使用datastore admin在google云存储中创建生产备份，然后在沙盒中使用datastore admin来恢复沙盒中的备份。在

将生产数据克隆到本地主机

我确实用一些生产数据来初始化本地主机数据存储，但这不是一个完整的克隆。只是核心需要的对象和一些测试用户。在

为此，我编写了一个google数据流作业，导出选定的模型并以jsonl格式保存在google云存储中。然后在我的本地主机上有一个名为/init/的端点，它启动一个taskqueue作业来下载和导入这些导出。在

为此，我重用了JSON REST处理程序代码，该代码能够将任何模型转换为JSON，反之亦然。在

理论上，您可以对整个数据存储执行此操作。在

编辑-这是my to json/from json代码的格式：

myndb.Model的所有子类myBaseModel具有通用转换代码：

get_dto_typemap = {
    ndb.DateTimeProperty: dt_to_timestamp,
    ndb.KeyProperty: key_to_dto,
    ndb.StringProperty: str_to_dto,
    ndb.EnumProperty: str,
}
set_from_dto_typemap = {
    ndb.DateTimeProperty: timestamp_to_dt,
    ndb.KeyProperty: dto_to_key,
    ndb.FloatProperty: float_from_dto,
    ndb.StringProperty: strip,
    ndb.BlobProperty: str,
    ndb.IntegerProperty: int,
}

class BaseModel(ndb.Model):

    def to_dto(self):
        dto = {'key': key_to_dto(self.key)}
        for name, obj in self._properties.iteritems():
            key = obj._name
            value = getattr(self, obj._name)
            if obj.__class__ in get_dto_typemap:
                if obj._repeated:
                    value = [get_dto_typemap[obj.__class__](v) for v in value]
                else:
                    value = get_dto_typemap[obj.__class__](value)
            dto[key] = value
        return dto

    def set_from_dto(self, dto):
        for name, obj in self._properties.iteritems():
            if isinstance(obj, ndb.ComputedProperty):
                continue
            key = obj._name
            if key in dto:
                value = dto[key]
                if not obj._repeated and obj.__class__ in set_from_dto_typemap:
                    try:
                        value = set_from_dto_typemap[obj.__class__](value)
                    except Exception as e:
                        raise Exception("Error setting "+self.__class__.__name__+"."+str(key)+" to '"+str(value) + "': " + e.message)
                try:
                    setattr(self, obj._name, value)
                except Exception as e:
                    print dir(obj)
                    raise Exception("Error setting "+self.__class__.__name__+"."+str(key)+" to '"+str(value)+"': "+e.message)

class User(BaseModel):
    # user fields, etc

然后，我的请求处理程序使用set_from_dto&；to_dto这样（BaseHandler还提供了一些将json有效负载转换为python dict的方便方法，等等）：

^{pr2}$