0

我正在尝试在 google appengine 数据存储表上开发行级访问。到目前为止，我确实有一个使用 _hooks 的常规 ndb put()、get() 和 delete() 操作的工作示例。

所有其他表都应使用 Acl 类。它用作结构化属性。

class Acl(EndpointsModel):
    UNAUTHORIZED_ERROR = 'Invalid token.'
    FORBIDDEN_ERROR = 'Permission denied.'

    public = ndb.BooleanProperty()
    readers = ndb.UserProperty(repeated=True)
    writers = ndb.UserProperty(repeated=True)
    owners = ndb.UserProperty(repeated=True)

    @classmethod
    def require_user(cls):
        current_user = endpoints.get_current_user()
        if current_user is None:
            raise endpoints.UnauthorizedException(cls.UNAUTHORIZED_ERROR)
        return current_user

    @classmethod
    def require_reader(cls, record):
        if not record:
            raise endpoints.NotFoundException(record.NOT_FOUND_ERROR)
        current_user = cls.require_user()
        if record.acl.public is not True or current_user not in record.acl.readers:
            raise endpoints.ForbiddenException(cls.FORBIDDEN_ERROR)

我确实想保护对 Location 类的访问。所以我确实在类中添加了三个钩子（_post_get_hook、_pre_put_hook 和 _pre_delete_hook）。

class Location(EndpointsModel):
    QUERY_FIELDS = ('state', 'limit', 'order', 'pageToken')
    NOT_FOUND_ERROR = 'Location not found.'

    description = ndb.TextProperty()
    address = ndb.StringProperty()
    acl = ndb.StructuredProperty(Acl)

    @classmethod
    def _post_get_hook(cls, key, future):
        location = future.get_result()
        Acl.require_reader(location)

    def _pre_put_hook(self):
        if self.key.id() is None:
            current_user = Acl.require_user()
            self.acl = Acl()
            self.acl.readers.append(current_user)
            self.acl.writers.append(current_user)
            self.acl.owners.append(current_user)
        else:
            location = self.key.get()
            Acl.require_writer(location)

这适用于所有创建、读取、更新和删除操作，但不适用于查询。

@Location.query_method(user_required=True,
                       path='location', http_method='GET', name='location.query')
def location_query(self, query):
    """
    Queries locations
    """
    current_user = Acl.require_user()
    query = query.filter(ndb.OR(Location.acl.readers == current_user, Location.acl.public == True))
    return query

当我对所有位置运行查询时，我收到以下错误消息：

BadArgumentError: _MultiQuery with cursors requires __key__ order

现在我有一些问题：

  • 如何解决 _MultiQuery 问题？
  • 一旦修复：这个 Acl 实现有意义吗？有开箱即用的替代品吗？（我想将 Acl 存储在记录本身上，以便能够运行直接查询，而不必先获取密钥。）
4

1 回答 1

2

Datastore 本身不支持OR过滤器。相反，NDB 在幕后所做的是运行两个查询：

 query.filter(Location.acl.readers == current_user)
 query.filter(Location.acl.public == True)

然后它将这两个查询的结果合并到一个结果集中。为了正确合并结果（特别是当您有重复的属性时消除重复），当从任意位置（使用游标）继续查询时，查询需要按键排序。

为了成功运行查询，您需要在运行之前将键顺序附加到查询：

def location_query(self, query):
"""
Queries locations
"""
current_user = Acl.require_user()
query = query.filter(ndb.OR(Location.acl.readers == current_user,
                            Location.acl.public == True)
                    ).order(Location.key)
return query

不幸的是，您的 ACL 实现不适用于查询。特别_post_get_hook是不要求查询结果。问题跟踪器上有一个关于此的错误。

于 2015-01-23T19:40:21.533 回答