| """Higher-level Query wrapper. |
| |
| There are perhaps too many query APIs in the world. |
| |
| The fundamental API here overloads the 6 comparisons operators to |
| represent filters on property values, and supports AND and OR |
| operations (implemented as functions -- Python's 'and' and 'or' |
| operators cannot be overloaded, and the '&' and '|' operators have a |
| priority that conflicts with the priority of comparison operators). |
| For example: |
| |
| class Employee(Model): |
| name = StringProperty() |
| age = IntegerProperty() |
| rank = IntegerProperty() |
| |
| @classmethod |
| def demographic(cls, min_age, max_age): |
| return cls.query().filter(AND(cls.age >= min_age, cls.age <= max_age)) |
| |
| @classmethod |
| def ranked(cls, rank): |
| return cls.query(cls.rank == rank).order(cls.age) |
| |
| for emp in Employee.seniors(42, 5): |
| print emp.name, emp.age, emp.rank |
| |
| The 'in' operator cannot be overloaded, but is supported through the |
| IN() method. For example: |
| |
| Employee.query().filter(Employee.rank.IN([4, 5, 6])) |
| |
| Sort orders are supported through the order() method; unary minus is |
| overloaded on the Property class to represent a descending order: |
| |
| Employee.query().order(Employee.name, -Employee.age) |
| |
| Besides using AND() and OR(), filters can also be combined by |
| repeatedly calling .filter(): |
| |
| q1 = Employee.query() # A query that returns all employees |
| q2 = q1.filter(Employee.age >= 30) # Only those over 30 |
| q3 = q2.filter(Employee.age < 40) # Only those in their 30s |
| |
| A further shortcut is calling .filter() with multiple arguments; this |
| implies AND(): |
| |
| q1 = Employee.query() # A query that returns all employees |
| q3 = q1.filter(Employee.age >= 30, |
| Employee.age < 40) # Only those in their 30s |
| |
| And finally you can also pass one or more filter expressions directly |
| to the .query() method: |
| |
| q3 = Employee.query(Employee.age >= 30, |
| Employee.age < 40) # Only those in their 30s |
| |
| Query objects are immutable, so these methods always return a new |
| Query object; the above calls to filter() do not affect q1. (On the |
| other hand, operations that are effectively no-ops may return the |
| original Query object.) |
| |
| Sort orders can also be combined this way, and .filter() and .order() |
| calls may be intermixed: |
| |
| q4 = q3.order(-Employee.age) |
| q5 = q4.order(Employee.name) |
| q6 = q5.filter(Employee.rank == 5) |
| |
| Again, multiple .order() calls can be combined: |
| |
| q5 = q3.order(-Employee.age, Employee.name) |
| |
| The simplest way to retrieve Query results is a for-loop: |
| |
| for emp in q3: |
| print emp.name, emp.age |
| |
| Some other methods to run a query and access its results: |
| |
| q.iter() # Return an iterator; same as iter(q) but more flexible |
| q.map(callback) # Call the callback function for each query result |
| q.fetch(N) # Return a list of the first N results |
| q.get() # Return the first result |
| q.count(N) # Return the number of results, with a maximum of N |
| q.fetch_page(N, start_cursor=cursor) # Return (results, cursor, has_more) |
| |
| All of the above methods take a standard set of additional query |
| options, either in the form of keyword arguments such as |
| keys_only=True, or as QueryOptions object passed with |
| options=QueryOptions(...). The most important query options are: |
| |
| keys_only: bool, if set the results are keys instead of entities |
| limit: int, limits the number of results returned |
| offset: int, skips this many results first |
| start_cursor: Cursor, start returning results after this position |
| end_cursor: Cursor, stop returning results after this position |
| batch_size: int, hint for the number of results returned per RPC |
| prefetch_size: int, hint for the number of results in the first RPC |
| produce_cursors: bool, return Cursor objects with the results |
| |
| For additional (obscure) query options and more details on them, |
| including an explanation of Cursors, see datastore_query.py. |
| |
| All of the above methods except for iter() have asynchronous variants |
| as well, which return a Future; to get the operation's ultimate |
| result, yield the Future (when inside a tasklet) or call the Future's |
| get_result() method (outside a tasklet): |
| |
| q.map_async(callback) # Callback may be a task or a plain function |
| q.fetch_async(N) |
| q.get_async() |
| q.count_async(N) |
| q.fetch_page_async(N, start_cursor=cursor) |
| |
| Finally, there's an idiom to efficiently loop over the Query results |
| in a tasklet, properly yielding when appropriate: |
| |
| it = q.iter() |
| while (yield it.has_next_async()): |
| emp = it.next() |
| print emp.name, emp.age |
| """ |
| |
| from __future__ import with_statement |
| del with_statement # No need to export this. |
| |
| __author__ = 'guido@google.com (Guido van Rossum)' |
| |
| import datetime |
| import heapq |
| import itertools |
| import sys |
| |
| from .google_imports import datastore_errors |
| from .google_imports import datastore_rpc |
| from .google_imports import datastore_types |
| from .google_imports import datastore_query |
| from .google_imports import namespace_manager |
| |
| from . import model |
| from . import context |
| from . import tasklets |
| from . import utils |
| |
| __all__ = ['Query', 'QueryOptions', 'Cursor', 'QueryIterator', |
| 'RepeatedStructuredPropertyPredicate', |
| 'AND', 'OR', 'ConjunctionNode', 'DisjunctionNode', |
| 'FilterNode', 'PostFilterNode', 'FalseNode', 'Node', |
| 'ParameterNode', 'ParameterizedThing', 'Parameter', |
| 'ParameterizedFunction', 'gql', |
| ] |
| |
| # Re-export some useful classes from the lower-level module. |
| Cursor = datastore_query.Cursor |
| |
| # Some local renamings. |
| _ASC = datastore_query.PropertyOrder.ASCENDING |
| _DESC = datastore_query.PropertyOrder.DESCENDING |
| _AND = datastore_query.CompositeFilter.AND |
| _KEY = datastore_types._KEY_SPECIAL_PROPERTY |
| |
| # Table of supported comparison operators. |
| _OPS = frozenset(['=', '!=', '<', '<=', '>', '>=', 'in']) |
| |
| # Default limit value. (Yes, the datastore uses int32!) |
| _MAX_LIMIT = 2 ** 31 - 1 |
| |
| |
| class QueryOptions(context.ContextOptions, datastore_query.QueryOptions): |
| """Support both context options and query options (esp. use_cache).""" |
| |
| |
| class RepeatedStructuredPropertyPredicate(datastore_query.FilterPredicate): |
| # Used by model.py. |
| |
| def __init__(self, match_keys, pb, key_prefix): |
| super(RepeatedStructuredPropertyPredicate, self).__init__() |
| self.match_keys = match_keys |
| stripped_keys = [] |
| for key in match_keys: |
| if not key.startswith(key_prefix): |
| raise ValueError('key %r does not begin with the specified prefix of %s' |
| % (key, key_prefix)) |
| stripped_keys.append(key[len(key_prefix):]) |
| value_map = datastore_query._make_key_value_map(pb, stripped_keys) |
| self.match_values = tuple(value_map[key][0] for key in stripped_keys) |
| |
| def _get_prop_names(self): |
| return frozenset(self.match_keys) |
| |
| def _apply(self, key_value_map): |
| """Apply the filter to values extracted from an entity. |
| |
| Think of self.match_keys and self.match_values as representing a |
| table with one row. For example: |
| |
| match_keys = ('name', 'age', 'rank') |
| match_values = ('Joe', 24, 5) |
| |
| (Except that in reality, the values are represented by tuples |
| produced by datastore_types.PropertyValueToKeyValue().) |
| |
| represents this table: |
| |
| | name | age | rank | |
| +---------+-------+--------+ |
| | 'Joe' | 24 | 5 | |
| |
| Think of key_value_map as a table with the same structure but |
| (potentially) many rows. This represents a repeated structured |
| property of a single entity. For example: |
| |
| {'name': ['Joe', 'Jane', 'Dick'], |
| 'age': [24, 21, 23], |
| 'rank': [5, 1, 2]} |
| |
| represents this table: |
| |
| | name | age | rank | |
| +---------+-------+--------+ |
| | 'Joe' | 24 | 5 | |
| | 'Jane' | 21 | 1 | |
| | 'Dick' | 23 | 2 | |
| |
| We must determine wheter at least one row of the second table |
| exactly matches the first table. We need this class because the |
| datastore, when asked to find an entity with name 'Joe', age 24 |
| and rank 5, will include entities that have 'Joe' somewhere in the |
| name column, 24 somewhere in the age column, and 5 somewhere in |
| the rank column, but not all aligned on a single row. Such an |
| entity should not be considered a match. |
| """ |
| columns = [] |
| for key in self.match_keys: |
| column = key_value_map.get(key) |
| if not column: # None, or an empty list. |
| return False # If any column is empty there can be no match. |
| columns.append(column) |
| # Use izip to transpose the columns into rows. |
| return self.match_values in itertools.izip(*columns) |
| |
| # Don't implement _prune()! It would mess up the row correspondence |
| # within columns. |
| |
| |
| class ParameterizedThing(object): |
| """Base class for Parameter and ParameterizedFunction. |
| |
| This exists purely for isinstance() checks. |
| """ |
| |
| def __eq__(self, other): |
| raise NotImplementedError |
| |
| def __ne__(self, other): |
| eq = self.__eq__(other) |
| if eq is not NotImplemented: |
| eq = not eq |
| return eq |
| |
| class Parameter(ParameterizedThing): |
| """Represents a bound variable in a GQL query. |
| |
| Parameter(1) corresponds to a slot labeled ":1" in a GQL query. |
| Parameter('xyz') corresponds to a slot labeled ":xyz". |
| |
| The value must be set (bound) separately by calling .set(value). |
| """ |
| |
| def __init__(self, key): |
| """Constructor. |
| |
| Args: |
| key: The Parameter key, must be either an integer or a string. |
| """ |
| if not isinstance(key, (int, long, basestring)): |
| raise TypeError('Parameter key must be an integer or string, not %s' % |
| (key,)) |
| self.__key = key |
| |
| def __repr__(self): |
| return '%s(%r)' % (self.__class__.__name__, self.__key) |
| |
| def __eq__(self, other): |
| if not isinstance(other, Parameter): |
| return NotImplemented |
| return self.__key == other.__key |
| |
| @property |
| def key(self): |
| """Retrieve the key.""" |
| return self.__key |
| |
| def resolve(self, bindings, used): |
| key = self.__key |
| if key not in bindings: |
| raise datastore_errors.BadArgumentError( |
| 'Parameter :%s is not bound.' % key) |
| value = bindings[key] |
| used[key] = True |
| return value |
| |
| |
| class ParameterizedFunction(ParameterizedThing): |
| """Represents a GQL function with parameterized arguments. |
| |
| For example, ParameterizedFunction('key', [Parameter(1)]) stands for |
| the GQL syntax KEY(:1). |
| """ |
| |
| def __init__(self, func, values): |
| from .google_imports import gql # Late import, to avoid name conflict. |
| self.__func = func |
| self.__values = values |
| # NOTE: A horrible hack using GQL private variables so we can |
| # reuse GQL's implementations of its built-in functions. |
| gqli = gql.GQL('SELECT * FROM Dummy') |
| gql_method = gqli._GQL__cast_operators[func] |
| self.__method = getattr(gqli, '_GQL' + gql_method.__name__) |
| |
| def __repr__(self): |
| return 'ParameterizedFunction(%r, %r)' % (self.__func, self.__values) |
| |
| def __eq__(self, other): |
| if not isinstance(other, ParameterizedFunction): |
| return NotImplemented |
| return (self.__func == other.__func and |
| self.__values == other.__values) |
| |
| @property |
| def func(self): |
| return self.__func |
| |
| @property |
| def values(self): |
| return self.__values |
| |
| def is_parameterized(self): |
| for val in self.__values: |
| if isinstance(val, Parameter): |
| return True |
| return False |
| |
| def resolve(self, bindings, used): |
| values = [] |
| for val in self.__values: |
| if isinstance(val, Parameter): |
| val = val.resolve(bindings, used) |
| values.append(val) |
| result = self.__method(values) |
| # The gql module returns slightly different types in some cases. |
| if self.__func == 'key' and isinstance(result, datastore_types.Key): |
| result = model.Key.from_old_key(result) |
| elif self.__func == 'time' and isinstance(result, datetime.datetime): |
| result = datetime.time(result.hour, result.minute, |
| result.second, result.microsecond) |
| elif self.__func == 'date' and isinstance(result, datetime.datetime): |
| result = datetime.date(result.year, result.month, result.day) |
| return result |
| |
| |
| class Node(object): |
| """Base class for filter expression tree nodes. |
| |
| Tree nodes are considered immutable, even though they can contain |
| Parameter instances, which are not. In particular, two identical |
| trees may be represented by the same Node object in different |
| contexts. |
| """ |
| |
| def __new__(cls): |
| if cls is Node: |
| raise TypeError('Cannot instantiate Node, only a subclass.') |
| return super(Node, cls).__new__(cls) |
| |
| def __eq__(self, other): |
| raise NotImplementedError |
| |
| def __ne__(self, other): |
| eq = self.__eq__(other) |
| if eq is not NotImplemented: |
| eq = not eq |
| return eq |
| |
| def __unordered(self, unused_other): |
| raise TypeError('Nodes cannot be ordered') |
| __le__ = __lt__ = __ge__ = __gt__ = __unordered |
| |
| def _to_filter(self, post=False): |
| """Helper to convert to datastore_query.Filter, or None.""" |
| raise NotImplementedError |
| |
| def _post_filters(self): |
| """Helper to extract post-filter Nodes, if any.""" |
| return None |
| |
| def resolve(self, bindings, used): |
| """Return a Node with Parameters replaced by the selected values. |
| |
| Args: |
| bindings: A dict mapping integers and strings to values. |
| used: A dict into which use of use of a binding is recorded. |
| |
| Returns: |
| A Node instance. |
| """ |
| return self |
| |
| |
| class FalseNode(Node): |
| """Tree node for an always-failing filter.""" |
| |
| def __eq__(self, other): |
| if not isinstance(other, FalseNode): |
| return NotImplemented |
| return True |
| |
| def _to_filter(self, post=False): |
| if post: |
| return None |
| # Because there's no point submitting a query that will never |
| # return anything. |
| raise datastore_errors.BadQueryError( |
| 'Cannot convert FalseNode to predicate') |
| |
| |
| class ParameterNode(Node): |
| """Tree node for a parameterized filter.""" |
| |
| def __new__(cls, prop, op, param): |
| if not isinstance(prop, model.Property): |
| raise TypeError('Expected a Property, got %r' % (prop,)) |
| if op not in _OPS: |
| raise TypeError('Expected a valid operator, got %r' % (op,)) |
| if not isinstance(param, ParameterizedThing): |
| raise TypeError('Expected a ParameterizedThing, got %r' % (param,)) |
| obj = super(ParameterNode, cls).__new__(cls) |
| obj.__prop = prop |
| obj.__op = op |
| obj.__param = param |
| return obj |
| |
| def __repr__(self): |
| return 'ParameterNode(%r, %r, %r)' % (self.__prop, self.__op, self.__param) |
| |
| def __eq__(self, other): |
| if not isinstance(other, ParameterNode): |
| return NotImplemented |
| return (self.__prop._name == other.__prop._name and |
| self.__op == other.__op and |
| self.__param == other.__param) |
| |
| def _to_filter(self, post=False): |
| raise datastore_errors.BadArgumentError( |
| 'Parameter :%s is not bound.' % (self.__param.key,)) |
| |
| def resolve(self, bindings, used): |
| value = self.__param.resolve(bindings, used) |
| if self.__op == 'in': |
| return self.__prop._IN(value) |
| else: |
| return self.__prop._comparison(self.__op, value) |
| |
| |
| class FilterNode(Node): |
| """Tree node for a single filter expression.""" |
| |
| def __new__(cls, name, opsymbol, value): |
| if isinstance(value, model.Key): |
| value = value.to_old_key() |
| if opsymbol == '!=': |
| n1 = FilterNode(name, '<', value) |
| n2 = FilterNode(name, '>', value) |
| return DisjunctionNode(n1, n2) |
| if opsymbol == 'in': |
| if not isinstance(value, (list, tuple, set, frozenset)): |
| raise TypeError('in expected a list, tuple or set of values; ' |
| 'received %r' % value) |
| nodes = [FilterNode(name, '=', v) for v in value] |
| if not nodes: |
| return FalseNode() |
| if len(nodes) == 1: |
| return nodes[0] |
| return DisjunctionNode(*nodes) |
| self = super(FilterNode, cls).__new__(cls) |
| self.__name = name |
| self.__opsymbol = opsymbol |
| self.__value = value |
| return self |
| |
| def __repr__(self): |
| return '%s(%r, %r, %r)' % (self.__class__.__name__, |
| self.__name, self.__opsymbol, self.__value) |
| |
| def __eq__(self, other): |
| if not isinstance(other, FilterNode): |
| return NotImplemented |
| # TODO: Should nodes with values that compare equal but have |
| # different types really be considered equal? IIUC the datastore |
| # doesn't consider 1 equal to 1.0 when it compares property values. |
| return (self.__name == other.__name and |
| self.__opsymbol == other.__opsymbol and |
| self.__value == other.__value) |
| |
| def _to_filter(self, post=False): |
| if post: |
| return None |
| if self.__opsymbol in ('!=', 'in'): |
| raise NotImplementedError('Inequality filters are not single filter ' |
| 'expressions and therefore cannot be converted ' |
| 'to a single filter (%r)' % self.__opsymbol) |
| value = self.__value |
| return datastore_query.make_filter(self.__name.decode('utf-8'), |
| self.__opsymbol, value) |
| |
| |
| class PostFilterNode(Node): |
| """Tree node representing an in-memory filtering operation. |
| |
| This is used to represent filters that cannot be executed by the |
| datastore, for example a query for a structured value. |
| """ |
| |
| def __new__(cls, predicate): |
| self = super(PostFilterNode, cls).__new__(cls) |
| self.predicate = predicate |
| return self |
| |
| def __repr__(self): |
| return '%s(%s)' % (self.__class__.__name__, self.predicate) |
| |
| def __eq__(self, other): |
| if not isinstance(other, PostFilterNode): |
| return NotImplemented |
| return self is other |
| |
| def _to_filter(self, post=False): |
| if post: |
| return self.predicate |
| else: |
| return None |
| |
| |
| class ConjunctionNode(Node): |
| """Tree node representing a Boolean AND operator on two or more nodes.""" |
| |
| def __new__(cls, *nodes): |
| if not nodes: |
| raise TypeError('ConjunctionNode() requires at least one node.') |
| elif len(nodes) == 1: |
| return nodes[0] |
| clauses = [[]] # Outer: Disjunction; inner: Conjunction. |
| # TODO: Remove duplicates? |
| for node in nodes: |
| if not isinstance(node, Node): |
| raise TypeError('ConjunctionNode() expects Node instances as arguments;' |
| ' received a non-Node instance %r' % node) |
| if isinstance(node, DisjunctionNode): |
| # Apply the distributive law: (X or Y) and (A or B) becomes |
| # (X and A) or (X and B) or (Y and A) or (Y and B). |
| new_clauses = [] |
| for clause in clauses: |
| for subnode in node: |
| new_clause = clause + [subnode] |
| new_clauses.append(new_clause) |
| clauses = new_clauses |
| elif isinstance(node, ConjunctionNode): |
| # Apply half of the distributive law: (X or Y) and A becomes |
| # (X and A) or (Y and A). |
| for clause in clauses: |
| clause.extend(node.__nodes) |
| else: |
| # Ditto. |
| for clause in clauses: |
| clause.append(node) |
| if not clauses: |
| return FalseNode() |
| if len(clauses) > 1: |
| return DisjunctionNode(*[ConjunctionNode(*clause) for clause in clauses]) |
| self = super(ConjunctionNode, cls).__new__(cls) |
| self.__nodes = clauses[0] |
| return self |
| |
| def __iter__(self): |
| return iter(self.__nodes) |
| |
| def __repr__(self): |
| return 'AND(%s)' % (', '.join(map(str, self.__nodes))) |
| |
| def __eq__(self, other): |
| if not isinstance(other, ConjunctionNode): |
| return NotImplemented |
| return self.__nodes == other.__nodes |
| |
| def _to_filter(self, post=False): |
| filters = filter(None, |
| (node._to_filter(post=post) |
| for node in self.__nodes |
| if isinstance(node, PostFilterNode) == post)) |
| if not filters: |
| return None |
| if len(filters) == 1: |
| return filters[0] |
| return datastore_query.CompositeFilter(_AND, filters) |
| |
| def _post_filters(self): |
| post_filters = [node for node in self.__nodes |
| if isinstance(node, PostFilterNode)] |
| if not post_filters: |
| return None |
| if len(post_filters) == 1: |
| return post_filters[0] |
| if post_filters == self.__nodes: |
| return self |
| return ConjunctionNode(*post_filters) |
| |
| def resolve(self, bindings, used): |
| nodes = [node.resolve(bindings, used) for node in self.__nodes] |
| if nodes == self.__nodes: |
| return self |
| return ConjunctionNode(*nodes) |
| |
| |
| class DisjunctionNode(Node): |
| """Tree node representing a Boolean OR operator on two or more nodes.""" |
| |
| def __new__(cls, *nodes): |
| if not nodes: |
| raise TypeError('DisjunctionNode() requires at least one node') |
| elif len(nodes) == 1: |
| return nodes[0] |
| self = super(DisjunctionNode, cls).__new__(cls) |
| self.__nodes = [] |
| # TODO: Remove duplicates? |
| for node in nodes: |
| if not isinstance(node, Node): |
| raise TypeError('DisjunctionNode() expects Node instances as arguments;' |
| ' received a non-Node instance %r' % node) |
| if isinstance(node, DisjunctionNode): |
| self.__nodes.extend(node.__nodes) |
| else: |
| self.__nodes.append(node) |
| return self |
| |
| def __iter__(self): |
| return iter(self.__nodes) |
| |
| def __repr__(self): |
| return 'OR(%s)' % (', '.join(map(str, self.__nodes))) |
| |
| def __eq__(self, other): |
| if not isinstance(other, DisjunctionNode): |
| return NotImplemented |
| return self.__nodes == other.__nodes |
| |
| def resolve(self, bindings, used): |
| nodes = [node.resolve(bindings, used) for node in self.__nodes] |
| if nodes == self.__nodes: |
| return self |
| return DisjunctionNode(*nodes) |
| |
| |
| # AND and OR are preferred aliases for these. |
| AND = ConjunctionNode |
| OR = DisjunctionNode |
| |
| |
| def _args_to_val(func, args): |
| """Helper for GQL parsing to extract values from GQL expressions. |
| |
| This can extract the value from a GQL literal, return a Parameter |
| for a GQL bound parameter (:1 or :foo), and interprets casts like |
| KEY(...) and plain lists of values like (1, 2, 3). |
| |
| Args: |
| func: A string indicating what kind of thing this is. |
| args: One or more GQL values, each integer, string, or GQL literal. |
| """ |
| from .google_imports import gql # Late import, to avoid name conflict. |
| vals = [] |
| for arg in args: |
| if isinstance(arg, (int, long, basestring)): |
| val = Parameter(arg) |
| elif isinstance(arg, gql.Literal): |
| val = arg.Get() |
| else: |
| raise TypeError('Unexpected arg (%r)' % arg) |
| vals.append(val) |
| if func == 'nop': |
| if len(vals) != 1: |
| raise TypeError('"nop" requires exactly one value') |
| return vals[0] # May be a Parameter |
| pfunc = ParameterizedFunction(func, vals) |
| if pfunc.is_parameterized(): |
| return pfunc |
| else: |
| return pfunc.resolve({}, {}) |
| |
| |
| def _get_prop_from_modelclass(modelclass, name): |
| """Helper for FQL parsing to turn a property name into a property object. |
| |
| Args: |
| modelclass: The model class specified in the query. |
| name: The property name. This may contain dots which indicate |
| sub-properties of structured properties. |
| |
| Returns: |
| A Property object. |
| |
| Raises: |
| KeyError if the property doesn't exist and the model clas doesn't |
| derive from Expando. |
| """ |
| if name == '__key__': |
| return modelclass._key |
| |
| parts = name.split('.') |
| part, more = parts[0], parts[1:] |
| prop = modelclass._properties.get(part) |
| if prop is None: |
| if issubclass(modelclass, model.Expando): |
| prop = model.GenericProperty(part) |
| else: |
| raise TypeError('Model %s has no property named %r' % |
| (modelclass._get_kind(), part)) |
| |
| while more: |
| part = more.pop(0) |
| if not isinstance(prop, model.StructuredProperty): |
| raise TypeError('Model %s has no property named %r' % |
| (modelclass._get_kind(), part)) |
| maybe = getattr(prop, part, None) |
| if isinstance(maybe, model.Property) and maybe._name == part: |
| prop = maybe |
| else: |
| maybe = prop._modelclass._properties.get(part) |
| if maybe is not None: |
| # Must get it this way to get the copy with the long name. |
| # (See StructuredProperty.__getattr__() for details.) |
| prop = getattr(prop, maybe._code_name) |
| else: |
| if issubclass(prop._modelclass, model.Expando) and not more: |
| prop = model.GenericProperty() |
| prop._name = name # Bypass the restriction on dots. |
| else: |
| raise KeyError('Model %s has no property named %r' % |
| (prop._modelclass._get_kind(), part)) |
| |
| return prop |
| |
| |
| class Query(object): |
| """Query object. |
| |
| Usually constructed by calling Model.query(). |
| |
| See module docstring for examples. |
| |
| Note that not all operations on Queries are supported by _MultiQuery |
| instances; the latter are generated as necessary when any of the |
| operators !=, IN or OR is used. |
| """ |
| |
| @utils.positional(1) |
| def __init__(self, kind=None, ancestor=None, filters=None, orders=None, |
| app=None, namespace=None, default_options=None, |
| projection=None, group_by=None): |
| """Constructor. |
| Args: |
| kind: Optional kind string. |
| ancestor: Optional ancestor Key. |
| filters: Optional Node representing a filter expression tree. |
| orders: Optional datastore_query.Order object. |
| app: Optional app id. |
| namespace: Optional namespace. |
| default_options: Optional QueryOptions object. |
| projection: Optional list or tuple of properties to project. |
| group_by: Optional list or tuple of properties to group by. |
| """ |
| # TODO(arfuller): Accept projection=Model.key to mean keys_only. |
| # TODO(arfuller): Consider adding incremental function |
| # group_by_property(*args) and project(*args, distinct=False). |
| |
| # Validating input. |
| if ancestor is not None: |
| if isinstance(ancestor, ParameterizedThing): |
| if isinstance(ancestor, ParameterizedFunction): |
| if ancestor.func != 'key': |
| raise TypeError('ancestor cannot be a GQL function other than KEY') |
| else: |
| if not isinstance(ancestor, model.Key): |
| raise TypeError('ancestor must be a Key; received %r' % (ancestor,)) |
| if not ancestor.id(): |
| raise ValueError('ancestor cannot be an incomplete key') |
| if app is not None: |
| if app != ancestor.app(): |
| raise TypeError('app/ancestor mismatch') |
| if namespace is None: |
| namespace = ancestor.namespace() |
| else: |
| if namespace != ancestor.namespace(): |
| raise TypeError('namespace/ancestor mismatch') |
| if filters is not None: |
| if not isinstance(filters, Node): |
| raise TypeError('filters must be a query Node or None; received %r' % |
| (filters,)) |
| if orders is not None: |
| if not isinstance(orders, datastore_query.Order): |
| raise TypeError('orders must be an Order instance or None; received %r' |
| % (orders,)) |
| if default_options is not None: |
| if not isinstance(default_options, datastore_rpc.BaseConfiguration): |
| raise TypeError('default_options must be a Configuration or None; ' |
| 'received %r' % (default_options,)) |
| if projection is not None: |
| if default_options.projection is not None: |
| raise TypeError('cannot use projection= and ' |
| 'default_options.projection at the same time') |
| if default_options.keys_only is not None: |
| raise TypeError('cannot use projection= and ' |
| 'default_options.keys_only at the same time') |
| |
| self.__kind = kind # String. |
| self.__ancestor = ancestor # Key. |
| self.__filters = filters # None or Node subclass. |
| self.__orders = orders # None or datastore_query.Order instance. |
| self.__app = app |
| self.__namespace = namespace |
| self.__default_options = default_options |
| |
| # Checked late as _check_properties depends on local state. |
| self.__projection = None |
| if projection is not None: |
| if not projection: |
| raise TypeError('projection argument cannot be empty') |
| if not isinstance(projection, (tuple, list)): |
| raise TypeError( |
| 'projection must be a tuple, list or None; received %r' % |
| (projection,)) |
| self._check_properties(self._to_property_names(projection)) |
| self.__projection = tuple(projection) |
| |
| self.__group_by = None |
| if group_by is not None: |
| if not group_by: |
| raise TypeError('group_by argument cannot be empty') |
| if not isinstance(group_by, (tuple, list)): |
| raise TypeError( |
| 'group_by must be a tuple, list or None; received %r' % (group_by,)) |
| self._check_properties(self._to_property_names(group_by)) |
| self.__group_by = tuple(group_by) |
| |
| def __repr__(self): |
| args = [] |
| if self.app is not None: |
| args.append('app=%r' % self.app) |
| if (self.namespace is not None and |
| self.namespace != namespace_manager.get_namespace()): |
| # Only show the namespace if set and not the current namespace. |
| # (This is similar to what Key.__repr__() does.) |
| args.append('namespace=%r' % self.namespace) |
| if self.kind is not None: |
| args.append('kind=%r' % self.kind) |
| if self.ancestor is not None: |
| args.append('ancestor=%r' % self.ancestor) |
| if self.filters is not None: |
| args.append('filters=%r' % self.filters) |
| if self.orders is not None: |
| # TODO: Format orders better. |
| args.append('orders=...') # PropertyOrder doesn't have a good repr(). |
| if self.projection: |
| args.append('projection=%r' % (self._to_property_names(self.projection))) |
| if self.group_by: |
| args.append('group_by=%r' % (self._to_property_names(self.group_by))) |
| if self.default_options is not None: |
| args.append('default_options=%r' % self.default_options) |
| return '%s(%s)' % (self.__class__.__name__, ', '.join(args)) |
| |
| def _fix_namespace(self): |
| """Internal helper to fix the namespace. |
| |
| This is called to ensure that for queries without an explicit |
| namespace, the namespace used by async calls is the one in effect |
| at the time the async call is made, not the one in effect when the |
| the request is actually generated. |
| """ |
| if self.namespace is not None: |
| return self |
| namespace = namespace_manager.get_namespace() |
| return self.__class__(kind=self.kind, ancestor=self.ancestor, |
| filters=self.filters, orders=self.orders, |
| app=self.app, namespace=namespace, |
| default_options=self.default_options, |
| projection=self.projection, group_by=self.group_by) |
| |
| def _get_query(self, connection): |
| self.bind() # Raises an exception if there are unbound parameters. |
| kind = self.kind |
| ancestor = self.ancestor |
| if ancestor is not None: |
| ancestor = connection.adapter.key_to_pb(ancestor) |
| filters = self.filters |
| post_filters = None |
| if filters is not None: |
| post_filters = filters._post_filters() |
| filters = filters._to_filter() |
| group_by = None |
| if self.group_by: |
| group_by = self._to_property_names(self.group_by) |
| dsquery = datastore_query.Query(app=self.app, |
| namespace=self.namespace, |
| kind=kind.decode('utf-8') if kind else None, |
| ancestor=ancestor, |
| filter_predicate=filters, |
| order=self.orders, |
| group_by=group_by) |
| if post_filters is not None: |
| dsquery = datastore_query._AugmentedQuery( |
| dsquery, |
| in_memory_filter=post_filters._to_filter(post=True)) |
| return dsquery |
| |
| @tasklets.tasklet |
| def run_to_queue(self, queue, conn, options=None, dsquery=None): |
| """Run this query, putting entities into the given queue.""" |
| try: |
| multiquery = self._maybe_multi_query() |
| if multiquery is not None: |
| yield multiquery.run_to_queue(queue, conn, options=options) |
| return |
| |
| if dsquery is None: |
| dsquery = self._get_query(conn) |
| rpc = dsquery.run_async(conn, options) |
| while rpc is not None: |
| batch = yield rpc |
| rpc = batch.next_batch_async(options) |
| for i, result in enumerate(batch.results): |
| queue.putq((batch, i, result)) |
| queue.complete() |
| |
| except GeneratorExit: |
| raise |
| except Exception: |
| if not queue.done(): |
| _, e, tb = sys.exc_info() |
| queue.set_exception(e, tb) |
| raise |
| |
| @tasklets.tasklet |
| def _run_to_list(self, results, options=None): |
| # Internal version of run_to_queue(), without a queue. |
| ctx = tasklets.get_context() |
| conn = ctx._conn |
| dsquery = self._get_query(conn) |
| rpc = dsquery.run_async(conn, options) |
| while rpc is not None: |
| batch = yield rpc |
| if (batch.skipped_results and |
| datastore_query.FetchOptions.offset(options)): |
| offset = options.offset - batch.skipped_results |
| options = datastore_query.FetchOptions(offset=offset, config=options) |
| rpc = batch.next_batch_async(options) |
| for result in batch.results: |
| result = ctx._update_cache_from_query_result(result, options) |
| if result is not None: |
| results.append(result) |
| |
| raise tasklets.Return(results) |
| |
| def _needs_multi_query(self): |
| filters = self.filters |
| return filters is not None and isinstance(filters, DisjunctionNode) |
| |
| def _maybe_multi_query(self): |
| if not self._needs_multi_query(): |
| return None |
| # Switch to a _MultiQuery. |
| filters = self.filters |
| subqueries = [] |
| for subfilter in filters: |
| subquery = self.__class__(kind=self.kind, ancestor=self.ancestor, |
| filters=subfilter, orders=self.orders, |
| app=self.app, namespace=self.namespace, |
| default_options=self.default_options, |
| projection=self.projection, |
| group_by=self.group_by) |
| subqueries.append(subquery) |
| return _MultiQuery(subqueries) |
| |
| @property |
| def kind(self): |
| """Accessor for the kind (a string or None).""" |
| return self.__kind |
| |
| @property |
| def ancestor(self): |
| """Accessor for the ancestor (a Key or None).""" |
| return self.__ancestor |
| |
| @property |
| def filters(self): |
| """Accessor for the filters (a Node or None).""" |
| return self.__filters |
| |
| @property |
| def orders(self): |
| """Accessor for the filters (a datastore_query.Order or None).""" |
| return self.__orders |
| |
| @property |
| def app(self): |
| """Accessor for the app (a string or None).""" |
| return self.__app |
| |
| @property |
| def namespace(self): |
| """Accessor for the namespace (a string or None).""" |
| return self.__namespace |
| |
| @property |
| def default_options(self): |
| """Accessor for the default_options (a QueryOptions instance or None).""" |
| return self.__default_options |
| |
| @property |
| def group_by(self): |
| """Accessor for the group by properties (a tuple instance or None).""" |
| return self.__group_by |
| |
| @property |
| def projection(self): |
| """Accessor for the projected properties (a tuple instance or None).""" |
| return self.__projection |
| |
| @property |
| def is_distinct(self): |
| """True if results are guaranteed to contain a unique set of property |
| values. |
| |
| This happens when every property in the group_by is also in the projection. |
| """ |
| return bool(self.__group_by and |
| set(self._to_property_names(self.__group_by)) <= |
| set(self._to_property_names(self.__projection))) |
| |
| def filter(self, *args): |
| """Return a new Query with additional filter(s) applied.""" |
| if not args: |
| return self |
| preds = [] |
| f = self.filters |
| if f: |
| preds.append(f) |
| for arg in args: |
| if not isinstance(arg, Node): |
| raise TypeError('Cannot filter a non-Node argument; received %r' % arg) |
| preds.append(arg) |
| if not preds: |
| pred = None |
| elif len(preds) == 1: |
| pred = preds[0] |
| else: |
| pred = ConjunctionNode(*preds) |
| return self.__class__(kind=self.kind, ancestor=self.ancestor, |
| filters=pred, orders=self.orders, |
| app=self.app, namespace=self.namespace, |
| default_options=self.default_options, |
| projection=self.projection, group_by=self.group_by) |
| |
| def order(self, *args): |
| """Return a new Query with additional sort order(s) applied.""" |
| # q.order(Employee.name, -Employee.age) |
| if not args: |
| return self |
| orders = [] |
| o = self.orders |
| if o: |
| orders.append(o) |
| for arg in args: |
| if isinstance(arg, model.Property): |
| orders.append(datastore_query.PropertyOrder(arg._name, _ASC)) |
| elif isinstance(arg, datastore_query.Order): |
| orders.append(arg) |
| else: |
| raise TypeError('order() expects a Property or query Order; ' |
| 'received %r' % arg) |
| if not orders: |
| orders = None |
| elif len(orders) == 1: |
| orders = orders[0] |
| else: |
| orders = datastore_query.CompositeOrder(orders) |
| return self.__class__(kind=self.kind, ancestor=self.ancestor, |
| filters=self.filters, orders=orders, |
| app=self.app, namespace=self.namespace, |
| default_options=self.default_options, |
| projection=self.projection, group_by=self.group_by) |
| |
| # Datastore API using the default context. |
| |
| def iter(self, **q_options): |
| """Construct an iterator over the query. |
| |
| Args: |
| **q_options: All query options keyword arguments are supported. |
| |
| Returns: |
| A QueryIterator object. |
| """ |
| self.bind() # Raises an exception if there are unbound parameters. |
| return QueryIterator(self, **q_options) |
| |
| __iter__ = iter |
| |
| @utils.positional(2) |
| def map(self, callback, pass_batch_into_callback=None, |
| merge_future=None, **q_options): |
| """Map a callback function or tasklet over the query results. |
| |
| Args: |
| callback: A function or tasklet to be applied to each result; see below. |
| merge_future: Optional Future subclass; see below. |
| **q_options: All query options keyword arguments are supported. |
| |
| Callback signature: The callback is normally called with an entity |
| as argument. However if keys_only=True is given, it is called |
| with a Key. Also, when pass_batch_into_callback is True, it is |
| called with three arguments: the current batch, the index within |
| the batch, and the entity or Key at that index. The callback can |
| return whatever it wants. If the callback is None, a trivial |
| callback is assumed that just returns the entity or key passed in |
| (ignoring produce_cursors). |
| |
| Optional merge future: The merge_future is an advanced argument |
| that can be used to override how the callback results are combined |
| into the overall map() return value. By default a list of |
| callback return values is produced. By substituting one of a |
| small number of specialized alternatives you can arrange |
| otherwise. See tasklets.MultiFuture for the default |
| implementation and a description of the protocol the merge_future |
| object must implement the default. Alternatives from the same |
| module include QueueFuture, SerialQueueFuture and ReducingFuture. |
| |
| Returns: |
| When the query has run to completion and all callbacks have |
| returned, map() returns a list of the results of all callbacks. |
| (But see 'optional merge future' above.) |
| """ |
| return self.map_async(callback, |
| pass_batch_into_callback=pass_batch_into_callback, |
| merge_future=merge_future, |
| **q_options).get_result() |
| |
| @utils.positional(2) |
| def map_async(self, callback, pass_batch_into_callback=None, |
| merge_future=None, **q_options): |
| """Map a callback function or tasklet over the query results. |
| |
| This is the asynchronous version of Query.map(). |
| """ |
| qry = self._fix_namespace() |
| return tasklets.get_context().map_query( |
| qry, |
| callback, |
| pass_batch_into_callback=pass_batch_into_callback, |
| options=self._make_options(q_options), |
| merge_future=merge_future) |
| |
| @utils.positional(2) |
| def fetch(self, limit=None, **q_options): |
| """Fetch a list of query results, up to a limit. |
| |
| Args: |
| limit: How many results to retrieve at most. |
| **q_options: All query options keyword arguments are supported. |
| |
| Returns: |
| A list of results. |
| """ |
| return self.fetch_async(limit, **q_options).get_result() |
| |
| @utils.positional(2) |
| def fetch_async(self, limit=None, **q_options): |
| """Fetch a list of query results, up to a limit. |
| |
| This is the asynchronous version of Query.fetch(). |
| """ |
| if limit is None: |
| default_options = self._make_options(q_options) |
| if default_options is not None and default_options.limit is not None: |
| limit = default_options.limit |
| else: |
| limit = _MAX_LIMIT |
| q_options['limit'] = limit |
| q_options.setdefault('batch_size', limit) |
| if self._needs_multi_query(): |
| return self.map_async(None, **q_options) |
| # Optimization using direct batches. |
| options = self._make_options(q_options) |
| qry = self._fix_namespace() |
| return qry._run_to_list([], options=options) |
| |
| def get(self, **q_options): |
| """Get the first query result, if any. |
| |
| This is similar to calling q.fetch(1) and returning the first item |
| of the list of results, if any, otherwise None. |
| |
| Args: |
| **q_options: All query options keyword arguments are supported. |
| |
| Returns: |
| A single result, or None if there are no results. |
| """ |
| return self.get_async(**q_options).get_result() |
| |
| def get_async(self, **q_options): |
| """Get the first query result, if any. |
| |
| This is the asynchronous version of Query.get(). |
| """ |
| qry = self._fix_namespace() |
| return qry._get_async(**q_options) |
| |
| @tasklets.tasklet |
| def _get_async(self, **q_options): |
| """Internal version of get_async().""" |
| res = yield self.fetch_async(1, **q_options) |
| if not res: |
| raise tasklets.Return(None) |
| raise tasklets.Return(res[0]) |
| |
| @utils.positional(2) |
| def count(self, limit=None, **q_options): |
| """Count the number of query results, up to a limit. |
| |
| This returns the same result as len(q.fetch(limit)) but more |
| efficiently. |
| |
| Note that you must pass a maximum value to limit the amount of |
| work done by the query. |
| |
| Args: |
| limit: How many results to count at most. |
| **q_options: All query options keyword arguments are supported. |
| |
| Returns: |
| """ |
| return self.count_async(limit, **q_options).get_result() |
| |
| @utils.positional(2) |
| def count_async(self, limit=None, **q_options): |
| """Count the number of query results, up to a limit. |
| |
| This is the asynchronous version of Query.count(). |
| """ |
| qry = self._fix_namespace() |
| return qry._count_async(limit=limit, **q_options) |
| |
| @tasklets.tasklet |
| def _count_async(self, limit=None, **q_options): |
| """Internal version of count_async().""" |
| # TODO: Support offset by incorporating it to the limit. |
| if 'offset' in q_options: |
| raise NotImplementedError('.count() and .count_async() do not support ' |
| 'offsets at present.') |
| if 'limit' in q_options: |
| raise TypeError('Cannot specify limit as a non-keyword argument and as a ' |
| 'keyword argument simultaneously.') |
| elif limit is None: |
| limit = _MAX_LIMIT |
| if self._needs_multi_query(): |
| # _MultiQuery does not support iterating over result batches, |
| # so just fetch results and count them. |
| # TODO: Use QueryIterator to avoid materializing the results list. |
| q_options.setdefault('batch_size', limit) |
| q_options.setdefault('keys_only', True) |
| results = yield self.fetch_async(limit, **q_options) |
| raise tasklets.Return(len(results)) |
| |
| # Issue a special query requesting 0 results at a given offset. |
| # The skipped_results count will tell us how many hits there were |
| # before that offset without fetching the items. |
| q_options['offset'] = limit |
| q_options['limit'] = 0 |
| options = self._make_options(q_options) |
| conn = tasklets.get_context()._conn |
| dsquery = self._get_query(conn) |
| rpc = dsquery.run_async(conn, options) |
| total = 0 |
| while rpc is not None: |
| batch = yield rpc |
| rpc = batch.next_batch_async(options) |
| total += batch.skipped_results |
| raise tasklets.Return(total) |
| |
| @utils.positional(2) |
| def fetch_page(self, page_size, **q_options): |
| """Fetch a page of results. |
| |
| This is a specialized method for use by paging user interfaces. |
| |
| Args: |
| page_size: The requested page size. At most this many results |
| will be returned. |
| |
| In addition, any keyword argument supported by the QueryOptions |
| class is supported. In particular, to fetch the next page, you |
| pass the cursor returned by one call to the next call using |
| start_cursor=<cursor>. A common idiom is to pass the cursor to |
| the client using <cursor>.to_websafe_string() and to reconstruct |
| that cursor on a subsequent request using |
| Cursor.from_websafe_string(<string>). |
| |
| Returns: |
| A tuple (results, cursor, more) where results is a list of query |
| results, cursor is a cursor pointing just after the last result |
| returned, and more is a bool indicating whether there are |
| (likely) more results after that. |
| """ |
| # NOTE: page_size can't be passed as a keyword. |
| return self.fetch_page_async(page_size, **q_options).get_result() |
| |
| @utils.positional(2) |
| def fetch_page_async(self, page_size, **q_options): |
| """Fetch a page of results. |
| |
| This is the asynchronous version of Query.fetch_page(). |
| """ |
| qry = self._fix_namespace() |
| return qry._fetch_page_async(page_size, **q_options) |
| |
| @tasklets.tasklet |
| def _fetch_page_async(self, page_size, **q_options): |
| """Internal version of fetch_page_async().""" |
| q_options.setdefault('batch_size', page_size) |
| q_options.setdefault('produce_cursors', True) |
| it = self.iter(limit=page_size + 1, **q_options) |
| results = [] |
| while (yield it.has_next_async()): |
| results.append(it.next()) |
| if len(results) >= page_size: |
| break |
| try: |
| cursor = it.cursor_after() |
| except datastore_errors.BadArgumentError: |
| cursor = None |
| raise tasklets.Return(results, cursor, it.probably_has_next()) |
| |
| def _make_options(self, q_options): |
| """Helper to construct a QueryOptions object from keyword arguments. |
| |
| Args: |
| q_options: a dict of keyword arguments. |
| |
| Note that either 'options' or 'config' can be used to pass another |
| QueryOptions object, but not both. If another QueryOptions object is |
| given it provides default values. |
| |
| If self.default_options is set, it is used to provide defaults, |
| which have a lower precedence than options set in q_options. |
| |
| Returns: |
| A QueryOptions object, or None if q_options is empty. |
| """ |
| if not (q_options or self.__projection): |
| return self.default_options |
| if 'options' in q_options: |
| # Move 'options' to 'config' since that is what QueryOptions() uses. |
| if 'config' in q_options: |
| raise TypeError('You cannot use config= and options= at the same time') |
| q_options['config'] = q_options.pop('options') |
| if q_options.get('projection'): |
| try: |
| q_options['projection'] = self._to_property_names( |
| q_options['projection']) |
| except TypeError, e: |
| raise datastore_errors.BadArgumentError(e) |
| self._check_properties(q_options['projection']) |
| options = QueryOptions(**q_options) |
| |
| # Populate projection if it hasn't been overridden. |
| if (options.keys_only is None and |
| options.projection is None and |
| self.__projection): |
| options = QueryOptions( |
| projection=self._to_property_names(self.__projection), config=options) |
| # Populate default options |
| if self.default_options is not None: |
| options = self.default_options.merge(options) |
| |
| return options |
| |
| def _to_property_names(self, properties): |
| if not isinstance(properties, (list, tuple)): |
| properties = [properties] # It will be type-checked below. |
| fixed = [] |
| for proj in properties: |
| if isinstance(proj, basestring): |
| fixed.append(proj) |
| elif isinstance(proj, model.Property): |
| fixed.append(proj._name) |
| else: |
| raise TypeError( |
| 'Unexpected property (%r); should be string or Property' % (proj,)) |
| return fixed |
| |
| def _check_properties(self, fixed, **kwargs): |
| modelclass = model.Model._kind_map.get(self.__kind) |
| if modelclass is not None: |
| modelclass._check_properties(fixed, **kwargs) |
| |
| def analyze(self): |
| """Return a list giving the parameters required by a query.""" |
| class MockBindings(dict): |
| def __contains__(self, key): |
| self[key] = None |
| return True |
| bindings = MockBindings() |
| used = {} |
| ancestor = self.ancestor |
| if isinstance(ancestor, ParameterizedThing): |
| ancestor = ancestor.resolve(bindings, used) |
| filters = self.filters |
| if filters is not None: |
| filters = filters.resolve(bindings, used) |
| return sorted(used) # Returns only the keys. |
| |
| def bind(self, *args, **kwds): |
| """Bind parameter values. Returns a new Query object.""" |
| return self._bind(args, kwds) |
| |
| def _bind(self, args, kwds): |
| """Bind parameter values. Returns a new Query object.""" |
| bindings = dict(kwds) |
| for i, arg in enumerate(args): |
| bindings[i + 1] = arg |
| used = {} |
| ancestor = self.ancestor |
| if isinstance(ancestor, ParameterizedThing): |
| ancestor = ancestor.resolve(bindings, used) |
| filters = self.filters |
| if filters is not None: |
| filters = filters.resolve(bindings, used) |
| unused = [] |
| for i in xrange(1, 1 + len(args)): |
| if i not in used: |
| unused.append(i) |
| if unused: |
| raise datastore_errors.BadArgumentError( |
| 'Positional arguments %s were given but not used.' % |
| ', '.join(str(i) for i in unused)) |
| return self.__class__(kind=self.kind, ancestor=ancestor, |
| filters=filters, orders=self.orders, |
| app=self.app, namespace=self.namespace, |
| default_options=self.default_options, |
| projection=self.projection, group_by=self.group_by) |
| |
| |
| def gql(query_string, *args, **kwds): |
| """Parse a GQL query string. |
| |
| Args: |
| query_string: Full GQL query, e.g. 'SELECT * FROM Kind WHERE prop = 1'. |
| *args, **kwds: If present, used to call bind(). |
| |
| Returns: |
| An instance of query_class. |
| """ |
| qry = _gql(query_string) |
| if args or kwds: |
| qry = qry._bind(args, kwds) |
| return qry |
| |
| |
| @utils.positional(1) |
| def _gql(query_string, query_class=Query): |
| """Parse a GQL query string (internal version). |
| |
| Args: |
| query_string: Full GQL query, e.g. 'SELECT * FROM Kind WHERE prop = 1'. |
| query_class: Optional class to use, default Query. |
| |
| Returns: |
| An instance of query_class. |
| """ |
| from .google_imports import gql # Late import, to avoid name conflict. |
| gql_qry = gql.GQL(query_string) |
| kind = gql_qry.kind() |
| if kind is None: |
| # The query must be lacking a "FROM <kind>" class. Let Expando |
| # stand in for the model class (it won't actually be used to |
| # construct the results). |
| modelclass = model.Expando |
| else: |
| modelclass = model.Model._kind_map.get(kind) |
| if modelclass is None: |
| # If the Adapter has a default model, use it; raise KindError otherwise. |
| ctx = tasklets.get_context() |
| modelclass = ctx._conn.adapter.default_model |
| if modelclass is None: |
| raise model.KindError( |
| "No model class found for kind %r. Did you forget to import it?" % |
| (kind,)) |
| else: |
| # Adjust kind to the model class's kind (for PolyModel). |
| kind = modelclass._get_kind() |
| ancestor = None |
| flt = gql_qry.filters() |
| filters = list(modelclass._default_filters()) |
| for name_op in sorted(flt): |
| name, op = name_op |
| values = flt[name_op] |
| op = op.lower() |
| if op == 'is' and name == gql.GQL._GQL__ANCESTOR: |
| if len(values) != 1: |
| raise ValueError('"is" requires exactly one value') |
| [(func, args)] = values |
| ancestor = _args_to_val(func, args) |
| continue |
| if op not in _OPS: |
| raise NotImplementedError('Operation %r is not supported.' % op) |
| for (func, args) in values: |
| val = _args_to_val(func, args) |
| prop = _get_prop_from_modelclass(modelclass, name) |
| if prop._name != name: |
| raise RuntimeError('Whoa! _get_prop_from_modelclass(%s, %r) ' |
| 'returned a property whose name is %r?!' % |
| (modelclass.__name__, name, prop._name)) |
| if isinstance(val, ParameterizedThing): |
| node = ParameterNode(prop, op, val) |
| elif op == 'in': |
| node = prop._IN(val) |
| else: |
| node = prop._comparison(op, val) |
| filters.append(node) |
| if filters: |
| filters = ConjunctionNode(*filters) |
| else: |
| filters = None |
| orders = _orderings_to_orders(gql_qry.orderings(), modelclass) |
| offset = gql_qry.offset() |
| limit = gql_qry.limit() |
| if limit < 0: |
| limit = None |
| keys_only = gql_qry._keys_only |
| if not keys_only: |
| keys_only = None |
| options = QueryOptions(offset=offset, limit=limit, keys_only=keys_only) |
| projection = gql_qry.projection() |
| if gql_qry.is_distinct(): |
| group_by = projection |
| else: |
| group_by = None |
| qry = query_class(kind=kind, |
| ancestor=ancestor, |
| filters=filters, |
| orders=orders, |
| default_options=options, |
| projection=projection, |
| group_by=group_by) |
| return qry |
| |
| |
| class QueryIterator(object): |
| """This iterator works both for synchronous and async callers! |
| |
| For synchronous callers, just use: |
| |
| for entity in Account.query(): |
| <use entity> |
| |
| Async callers use this idiom: |
| |
| it = iter(Account.query()) |
| while (yield it.has_next_async()): |
| entity = it.next() |
| <use entity> |
| |
| You can also use q.iter([options]) instead of iter(q); this allows |
| passing query options such as keys_only or produce_cursors. |
| |
| When keys_only is set, it.next() returns a key instead of an entity. |
| |
| When produce_cursors is set, the methods it.cursor_before() and |
| it.cursor_after() return Cursor objects corresponding to the query |
| position just before and after the item returned by it.next(). |
| Before it.next() is called for the first time, both raise an |
| exception. Once the loop is exhausted, both return the cursor after |
| the last item returned. Calling it.has_next() does not affect the |
| cursors; you must call it.next() before the cursors move. Note that |
| sometimes requesting a cursor requires a datastore roundtrip (but |
| not if you happen to request a cursor corresponding to a batch |
| boundary). If produce_cursors is not set, both methods always raise |
| an exception. |
| |
| Note that queries requiring in-memory merging of multiple queries |
| (i.e. queries using the IN, != or OR operators) do not support query |
| options. |
| """ |
| |
| # When produce_cursors is set, _lookahead collects (batch, index) |
| # pairs passed to _extended_callback(), and (_batch, _index) |
| # contain the info pertaining to the current item. |
| _lookahead = None |
| _batch = None |
| _index = None |
| |
| # Indicate the loop is exhausted. |
| _exhausted = False |
| |
| @utils.positional(2) |
| def __init__(self, query, **q_options): |
| """Constructor. Takes a Query and query options. |
| |
| This is normally called by Query.iter() or Query.__iter__(). |
| """ |
| ctx = tasklets.get_context() |
| options = query._make_options(q_options) |
| callback = self._extended_callback |
| self._iter = ctx.iter_query(query, |
| callback=callback, |
| pass_batch_into_callback=True, |
| options=options) |
| self._fut = None |
| |
| def _extended_callback(self, batch, index, ent): |
| if self._exhausted: |
| raise RuntimeError('QueryIterator is already exhausted') |
| # TODO: Make _lookup a deque. |
| if self._lookahead is None: |
| self._lookahead = [] |
| self._lookahead.append((batch, index)) |
| return ent |
| |
| def _consume_item(self): |
| if self._lookahead: |
| self._batch, self._index = self._lookahead.pop(0) |
| else: |
| self._batch = self._index = None |
| |
| def cursor_before(self): |
| """Return the cursor before the current item. |
| |
| You must pass a QueryOptions object with produce_cursors=True |
| for this to work. |
| |
| If there is no cursor or no current item, raise BadArgumentError. |
| Before next() has returned there is no cursor. Once the loop is |
| exhausted, this returns the cursor after the last item. |
| """ |
| if self._batch is None: |
| raise datastore_errors.BadArgumentError('There is no cursor currently') |
| # TODO: if cursor_after() was called for the previous item |
| # reuse that result instead of computing it from scratch. |
| # (Some cursor() calls make a datastore roundtrip.) |
| # TODO: reimplement the cursor() call to use NDB async I/O; |
| # perhaps even add async versions of cursor_before/after. |
| return self._batch.cursor(self._index + self._exhausted) |
| |
| def cursor_after(self): |
| """Return the cursor after the current item. |
| |
| You must pass a QueryOptions object with produce_cursors=True |
| for this to work. |
| |
| If there is no cursor or no current item, raise BadArgumentError. |
| Before next() has returned there is no cursor. Once the loop is |
| exhausted, this returns the cursor after the last item. |
| """ |
| if self._batch is None: |
| raise datastore_errors.BadArgumentError('There is no cursor currently') |
| return self._batch.cursor(self._index + 1) # TODO: inline this as async. |
| |
| def index_list(self): |
| """Return the list of indexes used for this query. |
| |
| This returns a list of index representations, where an index |
| representation is the same as what is returned by get_indexes(). |
| |
| Before the first result, the information is unavailable, and then |
| None is returned. This is not the same as an empty list -- the |
| empty list means that no index was used to execute the query. (In |
| the dev_appserver, an empty list may also mean that only built-in |
| indexes were used; metadata queries also return an empty list |
| here.) |
| |
| Proper use is as follows: |
| q = <modelclass>.query(<filters>) |
| i = q.iter() |
| try: |
| i.next() |
| except Stopiteration: |
| pass |
| indexes = i.index_list() |
| assert isinstance(indexes, list) |
| |
| Notes: |
| - Forcing produce_cursors=False makes this always return None. |
| - This always returns None for a multi-query. |
| """ |
| # TODO: Technically it is possible to implement this for |
| # multi-query by merging all the index lists from each subquery. |
| # Return None if the batch has no attribute index_list. |
| # This also applies when the batch itself is None. |
| return getattr(self._batch, 'index_list', None) |
| |
| def __iter__(self): |
| """Iterator protocol: get the iterator for this iterator, i.e. self.""" |
| return self |
| |
| def probably_has_next(self): |
| """Return whether a next item is (probably) available. |
| |
| This is not quite the same as has_next(), because when |
| produce_cursors is set, some shortcuts are possible. However, in |
| some cases (e.g. when the query has a post_filter) we can get a |
| false positive (returns True but next() will raise StopIteration). |
| There are no false negatives, if Batch.more_results doesn't lie. |
| """ |
| if self._lookahead: |
| return True |
| if self._batch is not None: |
| return self._batch.more_results |
| return self.has_next() |
| |
| def has_next(self): |
| """Return whether a next item is available. |
| |
| See the module docstring for the usage pattern. |
| """ |
| return self.has_next_async().get_result() |
| |
| @tasklets.tasklet |
| def has_next_async(self): |
| """Return a Future whose result will say whether a next item is available. |
| |
| See the module docstring for the usage pattern. |
| """ |
| if self._fut is None: |
| self._fut = self._iter.getq() |
| flag = True |
| try: |
| yield self._fut |
| except EOFError: |
| flag = False |
| raise tasklets.Return(flag) |
| |
| def next(self): |
| """Iterator protocol: get next item or raise StopIteration.""" |
| if self._fut is None: |
| self._fut = self._iter.getq() |
| try: |
| try: |
| ent = self._fut.get_result() |
| self._consume_item() |
| return ent |
| except EOFError: |
| self._exhausted = True |
| raise StopIteration |
| finally: |
| self._fut = None |
| |
| |
| class _SubQueryIteratorState(object): |
| """Helper class for _MultiQuery.""" |
| |
| def __init__(self, batch_i_entity, iterator, dsquery, orders): |
| batch, index, entity = batch_i_entity |
| self.batch = batch |
| self.index = index |
| self.entity = entity |
| self.iterator = iterator |
| self.dsquery = dsquery |
| self.orders = orders |
| |
| def __cmp__(self, other): |
| if not isinstance(other, _SubQueryIteratorState): |
| raise NotImplementedError('Can only compare _SubQueryIteratorState ' |
| 'instances to other _SubQueryIteratorState ' |
| 'instances; not %r' % other) |
| if not self.orders == other.orders: |
| raise NotImplementedError('Cannot compare _SubQueryIteratorStates with ' |
| 'differing orders (%r != %r)' % |
| (self.orders, other.orders)) |
| lhs = self.entity._orig_pb |
| rhs = other.entity._orig_pb |
| lhs_filter = self.dsquery._filter_predicate |
| rhs_filter = other.dsquery._filter_predicate |
| names = self.orders._get_prop_names() |
| # TODO: In some future version, there won't be a need to add the |
| # filters' names. |
| if lhs_filter is not None: |
| names |= lhs_filter._get_prop_names() |
| if rhs_filter is not None: |
| names |= rhs_filter._get_prop_names() |
| lhs_value_map = datastore_query._make_key_value_map(lhs, names) |
| rhs_value_map = datastore_query._make_key_value_map(rhs, names) |
| if lhs_filter is not None: |
| lhs_filter._prune(lhs_value_map) |
| if rhs_filter is not None: |
| rhs_filter._prune(rhs_value_map) |
| return self.orders._cmp(lhs_value_map, rhs_value_map) |
| |
| |
| class _MultiQuery(object): |
| """Helper class to run queries involving !=, IN or OR operators.""" |
| |
| # This is not instantiated by the user directly, but implicitly when |
| # iterating over a query with at least one filter using an IN, OR or |
| # != operator. Note that some options must be interpreted by |
| # _MultiQuery instead of passed to the underlying Queries' methods, |
| # e.g. offset (though not necessarily limit, and I'm not sure about |
| # cursors). |
| |
| # TODO: Need a way to specify the unification of two queries that |
| # are identical except one has an ancestor and the other doesn't. |
| # The HR datastore makes that a useful special case. |
| |
| def __init__(self, subqueries): |
| if not isinstance(subqueries, list): |
| raise TypeError('subqueries must be a list; received %r' % subqueries) |
| for subq in subqueries: |
| if not isinstance(subq, Query): |
| raise TypeError('Each subquery must be a Query instances; received %r' |
| % subq) |
| first_subquery = subqueries[0] |
| kind = first_subquery.kind |
| orders = first_subquery.orders |
| if not kind: |
| raise ValueError('Subquery kind cannot be missing') |
| for subq in subqueries[1:]: |
| if subq.kind != kind: |
| raise ValueError('Subqueries must be for a common kind (%s != %s)' % |
| (subq.kind, kind)) |
| elif subq.orders != orders: |
| raise ValueError('Subqueries must have the same order(s) (%s != %s)' % |
| (subq.orders, orders)) |
| # TODO: Ensure that app and namespace match, when we support them. |
| self.__subqueries = subqueries |
| self.__orders = orders |
| self.ancestor = None # Hack for map_query(). |
| |
| def _make_options(self, q_options): |
| return self.__subqueries[0].default_options |
| |
| @property |
| def orders(self): |
| return self.__orders |
| |
| @property |
| def default_options(self): |
| return self.__subqueries[0].default_options |
| |
| @tasklets.tasklet |
| def run_to_queue(self, queue, conn, options=None): |
| """Run this query, putting entities into the given queue.""" |
| if options is None: |
| # Default options. |
| offset = None |
| limit = None |
| keys_only = None |
| else: |
| # Capture options we need to simulate. |
| offset = options.offset |
| limit = options.limit |
| keys_only = options.keys_only |
| |
| # Cursors are supported for certain orders only. |
| if (options.start_cursor or options.end_cursor or |
| options.produce_cursors): |
| names = set() |
| if self.__orders is not None: |
| names = self.__orders._get_prop_names() |
| if '__key__' not in names: |
| raise datastore_errors.BadArgumentError( |
| '_MultiQuery with cursors requires __key__ order') |
| |
| # Decide if we need to modify the options passed to subqueries. |
| # NOTE: It would seem we can sometimes let the datastore handle |
| # the offset natively, but this would thwart the duplicate key |
| # detection, so we always have to emulate the offset here. |
| # We can set the limit we pass along to offset + limit though, |
| # since that is the maximum number of results from a single |
| # subquery we will ever have to consider. |
| modifiers = {} |
| if offset: |
| modifiers['offset'] = None |
| if limit is not None: |
| modifiers['limit'] = min(_MAX_LIMIT, offset + limit) |
| if keys_only and self.__orders is not None: |
| modifiers['keys_only'] = None |
| if modifiers: |
| options = QueryOptions(config=options, **modifiers) |
| |
| if offset is None: |
| offset = 0 |
| |
| if limit is None: |
| limit = _MAX_LIMIT |
| |
| if self.__orders is None: |
| # Run the subqueries sequentially; there is no order to keep. |
| keys_seen = set() |
| for subq in self.__subqueries: |
| if limit <= 0: |
| break |
| subit = tasklets.SerialQueueFuture('_MultiQuery.run_to_queue[ser]') |
| subq.run_to_queue(subit, conn, options=options) |
| while limit > 0: |
| try: |
| batch, index, result = yield subit.getq() |
| except EOFError: |
| break |
| if keys_only: |
| key = result |
| else: |
| key = result._key |
| if key not in keys_seen: |
| keys_seen.add(key) |
| if offset > 0: |
| offset -= 1 |
| else: |
| limit -= 1 |
| queue.putq((None, None, result)) |
| queue.complete() |
| return |
| |
| # This with-statement causes the adapter to set _orig_pb on all |
| # entities it converts from protobuf. |
| # TODO: Does this interact properly with the cache? |
| with conn.adapter: |
| # Start running all the sub-queries. |
| todo = [] # List of (subit, dsquery) tuples. |
| for subq in self.__subqueries: |
| dsquery = subq._get_query(conn) |
| subit = tasklets.SerialQueueFuture('_MultiQuery.run_to_queue[par]') |
| subq.run_to_queue(subit, conn, options=options, dsquery=dsquery) |
| todo.append((subit, dsquery)) |
| |
| # Create a list of (first-entity, subquery-iterator) tuples. |
| state = [] # List of _SubQueryIteratorState instances. |
| for subit, dsquery in todo: |
| try: |
| thing = yield subit.getq() |
| except EOFError: |
| continue |
| else: |
| state.append(_SubQueryIteratorState(thing, subit, dsquery, |
| self.__orders)) |
| |
| # Now turn it into a sorted heap. The heapq module claims that |
| # calling heapify() is more efficient than calling heappush() for |
| # each item. |
| heapq.heapify(state) |
| |
| # Repeatedly yield the lowest entity from the state vector, |
| # filtering duplicates. This is essentially a multi-way merge |
| # sort. One would think it should be possible to filter |
| # duplicates simply by dropping other entities already in the |
| # state vector that are equal to the lowest entity, but because of |
| # the weird sorting of repeated properties, we have to explicitly |
| # keep a set of all keys, so we can remove later occurrences. |
| # Note that entities will still be sorted correctly, within the |
| # constraints given by the sort order. |
| keys_seen = set() |
| while state and limit > 0: |
| item = heapq.heappop(state) |
| batch = item.batch |
| index = item.index |
| entity = item.entity |
| key = entity._key |
| if key not in keys_seen: |
| keys_seen.add(key) |
| if offset > 0: |
| offset -= 1 |
| else: |
| limit -= 1 |
| if keys_only: |
| queue.putq((batch, index, key)) |
| else: |
| queue.putq((batch, index, entity)) |
| subit = item.iterator |
| try: |
| batch, index, entity = yield subit.getq() |
| except EOFError: |
| pass |
| else: |
| item.batch = batch |
| item.index = index |
| item.entity = entity |
| heapq.heappush(state, item) |
| queue.complete() |
| |
| # Datastore API using the default context. |
| |
| def iter(self, **q_options): |
| return QueryIterator(self, **q_options) |
| |
| __iter__ = iter |
| |
| # TODO: Add fetch() etc.? |
| |
| |
| # Helper functions to convert between orders and orderings. An order |
| # is a datastore_query.Order instance. An ordering is a |
| # (property_name, direction) tuple. |
| |
| def _order_to_ordering(order): |
| pb = order._to_pb() |
| return pb.property(), pb.direction() # TODO: What about UTF-8? |
| |
| |
| def _orders_to_orderings(orders): |
| if orders is None: |
| return [] |
| if isinstance(orders, datastore_query.PropertyOrder): |
| return [_order_to_ordering(orders)] |
| if isinstance(orders, datastore_query.CompositeOrder): |
| # TODO: What about UTF-8? |
| return [(pb.property(), pb.direction()) for pb in orders._to_pbs()] |
| raise ValueError('Bad order: %r' % (orders,)) |
| |
| |
| def _ordering_to_order(ordering, modelclass): |
| name, direction = ordering |
| prop = _get_prop_from_modelclass(modelclass, name) |
| if prop._name != name: |
| raise RuntimeError('Whoa! _get_prop_from_modelclass(%s, %r) ' |
| 'returned a property whose name is %r?!' % |
| (modelclass.__name__, name, prop._name)) |
| return datastore_query.PropertyOrder(name, direction) |
| |
| |
| def _orderings_to_orders(orderings, modelclass): |
| orders = [_ordering_to_order(o, modelclass) for o in orderings] |
| if not orders: |
| return None |
| if len(orders) == 1: |
| return orders[0] |
| return datastore_query.CompositeOrder(orders) |
