AppEngine/NDB How to apply a query filter against single entity

Question

In our app we often have several query filters that check things like archive status, permission, entity hierarchy, etc. These filters are built using NDB's operator overloads, e.g. models.Item.isArchived == False. I believe these are also referred to as FilterNode instances.

Given a set of these FilterNodes how can I apply them to a single entity that has been looked up outside of a query? Say from Memcache.

My use case is I have a predefined set of filters for an endpoint and the service allows the user to specify an entity ID. Rather than run a query, I do a get by ID. But I need to post filter to sure it's valid to return.

Answer 1

Found a couple of ways to do in-memory filtering using a Query object (both of these assume NDB... but can be modified for non)

this first one just uses Python builtin "filter()"... it doesn't handle sort-orders that may be in the query:

def filter_entities(query, entities):
  #convert entities to list of protocol buffers
  entity_pbs = [ent._to_pb() for ent in entities if ent is not None]

  #assuming all entities are same type... just get type from first one
  model_klass = type(entities[0])

  #turn the query into a filter and hand it to builtin
  filtered = filter(query.filters._to_filter(), entity_pbs)

  #convert protocol buffers back into entities and hand back
  return [klass._from_pb(ent) for ent in filtered if ent is not None]

This other approach uses apply_query from datastore_query... which will take sort orders into account:

from google.appengine.ext.ndb import tasklets
from google.appengine.datastore.datastore_query import apply_query

def apply_query(query, entities):
  #convert list of entities to list of protocol buffers
  entity_pbs = [ent._to_pb() for ent in entities if ent is not None]

  #convert NDB query to datastore_query (needs connection)
  conn = tasklets.get_context()._conn
  dsquery = query._get_query(conn)

  #call datastore_query.apply_query
  filtered = apply_query(dsquery, [entity_pbs])

  #convert protocol buffers back into entities and hand back
  return [klass._from_pb(ent) for ent in filtered if ent is not None]

clearly I'm accessing some hidden methods that could change in the future... and so be sure you've written tests to catch that possibility