如何将 NDB 数据传递到任务队列？

Posted 2023-02-16

【中文标题】如何将 NDB 数据传递到任务队列？

【英文标题】：How to pass NDB data to task queue?

【发布时间】：2014-05-30 15:09:14

【问题描述】：

这是我想传递给任务的实体：

class MyData(ndb.Model):
    ...
    text = ndb.StringProperty(indexed=False)
    data = ndb.BlobKeyProperty(repeated=True)
    details = ndb.KeyProperty(kind=Details)

我可以执行以下操作吗？

mydata = MyData.query()
mydata = mydata.filter(...)
mydata = mydata.order(MyData.added)
mydata = mydata.fetch(100)
for d in mydata:
  taskqueue.add(url='/worker', payload=d)

那我怎样才能从有效载荷中提取数据呢？不要认为self.request.get('payload') 会起作用。 了解我可以只传递 ndb 键并读取任务中的实体。但这将需要额外的读取操作。或者，我可以在fetch(100) 时以某种方式使用keys_only 吗？ keys_only按照the doc免费操作：

小型数据存储操作包括调用分配数据存储 ID 或 仅键查询，并且这些操作是免费的。

但它们是否算作数据存储读取操作？

【参考方案1】：

我会使用 keys_only 查询来执行此操作（正如您所提到的，这应该不会产生任何费用），并且我会更新您的任务队列以添加到单个批处理调用中。

mydata = MyData.query()
mydata = mydata.filter(...)
# the order shouldn't matter, unless you want to make sure that property exists.
# mydata = mydata.order(MyData.added)   
mydata = mydata.fetch(100, keys_only=True)
tasks = [taskqueue.Task(url='/worker', params='key': key.urlsafe()) for key in mydata]
taskqueue.Queue('default').add(tasks)

【讨论】：

谢谢！单批调用有什么好处？我打算为每个任务使用唯一的名称（这样，我不会对相同的错误进行重复处理），如果有重复名称的任务，批处理调用会发生什么？

能否请您帮助了解如何提取有效载荷？ self.request.get('payload') 为空。

将payload 更新为params - 现在您将使用self.request.get('key')。

developers.google.com/appengine/docs/python/taskqueue/… 描述了当您添加已存在的任务时会发生什么。如果您担心每个项目的重复任务，我会一次添加一个并捕获(taskqueue.TaskAlreadyExistsError, taskqueue.TombstonedTaskError)

啊，是的，我知道如何使用params，但对payload 尤其感兴趣;)

【参考方案2】：

d 仍然是一个 ndb 对象。要将其作为字典传递，试试这个（未经测试）：

taskqueue.add(url='/worker', payload=d.to_dict())

https://developers.google.com/appengine/docs/python/ndb/modelclass#Model_to_dict

【讨论】：

这将不起作用，因为 details 是一个 ndb.KeyProperty，如果没有一些额外的工作，它是不可 JSON 编码的。我会覆盖 to_dict 来处理 KeyProperty 并确保应用 .urlsafe() 将它们转换为字符串。然后，您以后获取这些实体就没有问题了。