Google appengine 데이터 저장소 쿼리에 대한 행 수준 액세스

Question

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) 

위치 클래스에 대한 액세스를 보호하고 싶습니다. 그래서 세 개의 후크 (_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을 저장하려고했습니다.)

Answer 1

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은 쿼리 결과에 대해 호출되지 않습니다. bug filed on the issue tracker about this이 있습니다.