我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用google.appengine.ext.db.Query()。
def make_directed_datastore_query(self, kind, keys_only=False): """Construct a query for this key range, including the scan direction. Args: kind: A string. keys_only: bool, default False, use keys_only on Query? Returns: A datastore.Query instance. Raises: KeyRangeError: if self.direction is not in (KeyRange.ASC, KeyRange.DESC). """ direction = self.__get_direction(datastore.Query.ASCENDING, datastore.Query.DESCENDING) query = datastore.Query(kind, _app=self._app, keys_only=keys_only) query.Order(("__key__", direction)) query = self.filter_datastore_query(query) return query
def make_ascending_query(self, kind_class, keys_only=False, filters=None): """Construct a query for this key range without setting the scan direction. Args: kind_class: A kind implementation class (a subclass of either db.Model or ndb.Model). keys_only: bool, default False, query only for keys. filters: optional list of filters to apply to the query. Each filter is a tuple: (<property_name_as_str>, <query_operation_as_str>, <value>). User filters are applied first. Returns: A db.Query or ndb.Query instance (corresponding to kind_class). """ if ndb is not None: if issubclass(kind_class, ndb.Model): return self.make_ascending_ndb_query( kind_class, keys_only=keys_only, filters=filters) assert self._app is None, '_app is not supported for db.Query' query = db.Query(kind_class, namespace=self.namespace, keys_only=keys_only) query.order("__key__") query = self.filter_query(query, filters=filters) return query
def make_ascending_ndb_query(self, kind_class, keys_only=False, filters=None): """Construct an NDB query for this key range, without the scan direction. Args: kind_class: An ndb.Model subclass. keys_only: bool, default False, query only for keys. Returns: An ndb.Query instance. """ assert issubclass(kind_class, ndb.Model) if keys_only: default_options = ndb.QueryOptions(keys_only=True) else: default_options = None query = kind_class.query(app=self._app, namespace=self.namespace, default_options=default_options) query = self.filter_ndb_query(query, filters=filters) query = query.order(kind_class._key) return query
def make_ascending_datastore_query(self, kind, keys_only=False, filters=None): """Construct a query for this key range without setting the scan direction. Args: kind: A string. keys_only: bool, default False, use keys_only on Query? filters: optional list of filters to apply to the query. Each filter is a tuple: (<property_name_as_str>, <query_operation_as_str>, <value>). User filters are applied first. Returns: A datastore.Query instance. """ query = datastore.Query(kind, namespace=self.namespace, _app=self._app, keys_only=keys_only) query.Order(("__key__", datastore.Query.ASCENDING)) query = self.filter_datastore_query(query, filters=filters) return query
def expand(self,expression,field_type=None): if isinstance(expression,Field): if expression.type in ('text', 'blob', 'json'): raise SyntaxError('AppEngine does not index by: %s' % expression.type) return expression.name elif isinstance(expression, (Expression, Query)): if not expression.second is None: return expression.op(expression.first, expression.second) elif not expression.first is None: return expression.op(expression.first) else: return expression.op() elif field_type: return self.represent(expression,field_type) elif isinstance(expression,(list,tuple)): return ','.join([self.represent(item,field_type) for item in expression]) else: return str(expression) ### TODO from gql.py Expression
def delete(self,tablename,query): if not isinstance(query,Query): raise SyntaxError("Not Supported") if query.first.type=='id' and query.op==self.EQ: id = query.second tablename = query.first.tablename assert(tablename == query.first.tablename) ctable = self.connection[tablename] try: del ctable[str(id)] return 1 except couchdb.http.ResourceNotFound: return 0 else: tablename = self.get_table(query) rows = self.select(query,[self.db[tablename]._id],{}) ctable = self.connection[tablename] for row in rows: del ctable[str(row.id)] return len(rows)
def update(self, tablename, query, fields, safe=None): if safe == None: safe = self.safe # return amount of adjusted rows or zero, but no exceptions # @ related not finding the result if not isinstance(query, Query): raise RuntimeError("Not implemented") amount = self.count(query, False) modify, filter = self._update(tablename, query, fields) try: result = self.connection[tablename].update(filter, modify, multi=True, safe=safe) if safe: try: # if result count is available fetch it return result["n"] except (KeyError, AttributeError, TypeError): return amount else: return amount except Exception, e: # TODO Reverse update query to verifiy that the query succeded raise RuntimeError("uncaught exception when updating rows: %s" % e)
def get_query_mailbox(self, query): nofield = True tablename = None attr = query while nofield: if hasattr(attr, "first"): attr = attr.first if isinstance(attr, Field): return attr.tablename elif isinstance(attr, Query): pass else: return None else: return None return tablename
def belongs(self, *value): """ Accepts the following inputs: field.belongs(1,2) field.belongs((1,2)) field.belongs(query) Does NOT accept: field.belongs(1) """ db = self.db if len(value) == 1: value = value[0] if isinstance(value,Query): value = db(value)._select(value.first._table._id) return Query(db, db._adapter.BELONGS, self, value)
def load(cls, backup_id, kind_name, shard_id=None): """Retrieve SchemaAggregationResult from the Datastore. Args: backup_id: Required BackupInformation Key. kind_name: Required kind name as string. shard_id: Optional shard id as string. Returns: SchemaAggregationResult iterator or an entity if shard_id not None. """ parent = cls._get_parent_key(backup_id, kind_name) if shard_id: key = datastore_types.Key.from_path(cls.kind(), shard_id, parent=parent) return SchemaAggregationResult.get(key) else: return db.Query(cls).ancestor(parent).run()
def make_directed_query(self, kind_class, keys_only=False): """Construct a query for this key range, including the scan direction. Args: kind_class: A kind implementation class (a subclass of either db.Model or ndb.Model). keys_only: bool, default False, use keys_only on Query? Returns: A db.Query or ndb.Query instance (corresponding to kind_class). Raises: KeyRangeError: if self.direction is not in (KeyRange.ASC, KeyRange.DESC). """ if ndb is not None: if issubclass(kind_class, ndb.Model): return self.make_directed_ndb_query(kind_class, keys_only=keys_only) assert self._app is None, '_app is not supported for db.Query' direction = self.__get_direction("", "-") query = db.Query(kind_class, namespace=self.namespace, keys_only=keys_only) query.order("%s__key__" % direction) query = self.filter_query(query) return query
def to_json(self): """Inherit doc.""" cursor = self._cursor if self._query is not None: if isinstance(self._query, db.Query): cursor = self._query.cursor() else: cursor = self._query.cursor_after() if cursor is None or isinstance(cursor, basestring): cursor_object = False else: cursor_object = True cursor = cursor.to_websafe_string() return {"property_range": self._property_range.to_json(), "query_spec": self._query_spec.to_json(), "cursor": cursor, "ns_range": self._ns_range.to_json_object(), "name": self.__class__.__name__, "cursor_object": cursor_object}
def __iter__(self): self._query = self._key_range.make_ascending_query( util.for_name(self._query_spec.model_class_path), filters=self._query_spec.filters) if isinstance(self._query, db.Query): if self._cursor: self._query.with_cursor(self._cursor) for model_instance in self._query.run( batch_size=self._query_spec.batch_size, keys_only=self._query_spec.keys_only): yield model_instance else: self._query = self._query.iter(batch_size=self._query_spec.batch_size, keys_only=self._query_spec.keys_only, start_cursor=self._cursor, produce_cursors=True) for model_instance in self._query: yield model_instance
def expand(self, expression, field_type=None): if isinstance(expression, Field): if expression.type in ('text', 'blob', 'json'): raise SyntaxError('AppEngine does not index by: %s' % expression.type) return expression.name elif isinstance(expression, (Expression, Query)): if not expression.second is None: return expression.op(expression.first, expression.second) elif not expression.first is None: return expression.op(expression.first) else: return expression.op() elif field_type: return self.represent(expression, field_type) elif isinstance(expression, (list, tuple)): return ','.join([self.represent(item, field_type) for item in expression]) else: return str(expression) ### TODO from gql.py Expression
def delete(self, tablename, query): if not isinstance(query, Query): raise SyntaxError("Not Supported") if query.first.type == 'id' and query.op == self.EQ: id = query.second tablename = query.first.tablename assert(tablename == query.first.tablename) ctable = self.connection[tablename] try: del ctable[str(id)] return 1 except couchdb.http.ResourceNotFound: return 0 else: tablename = self.get_table(query) rows = self.select(query, [self.db[tablename]._id], {}) ctable = self.connection[tablename] for row in rows: del ctable[str(row.id)] return len(rows)
def test_Query_type(self): """ L{db.Query} instances get converted to lists .. """ q = models.EmptyModel.all() self.assertTrue(isinstance(q, db.Query)) self.assertEncodes(q, b'\n\x00\x00\x00\x00', encoding=pyamf.AMF0) self.assertEncodes(q, b'\t\x01\x01', encoding=pyamf.AMF3)
def all(cls): """Get query for all Blobs associated with application. Returns: A db.Query object querying over BlobInfo's datastore kind. """ return db.Query(model_class=cls, namespace='')
def _IsNdbQuery(query): return ndb is not None and isinstance(query, ndb.Query)
def filter_query(self, query, filters=None): """Add query filter to restrict to this key range. Args: query: A db.Query or ndb.Query instance. filters: optional list of filters to apply to the query. Each filter is a tuple: (<property_name_as_str>, <query_operation_as_str>, <value>). User filters are applied first. Returns: The input query restricted to this key range. """ if ndb is not None: if _IsNdbQuery(query): return self.filter_ndb_query(query, filters=filters) assert not _IsNdbQuery(query) if filters: for f in filters: query.filter("%s %s" % (f[0], f[1]), f[2]) if self.include_start: start_comparator = ">=" else: start_comparator = ">" if self.include_end: end_comparator = "<=" else: end_comparator = "<" if self.key_start: query.filter("__key__ %s" % start_comparator, self.key_start) if self.key_end: query.filter("__key__ %s" % end_comparator, self.key_end) return query
def filter_ndb_query(self, query, filters=None): """Add query filter to restrict to this key range. Args: query: An ndb.Query instance. filters: optional list of filters to apply to the query. Each filter is a tuple: (<property_name_as_str>, <query_operation_as_str>, <value>). User filters are applied first. Returns: The input query restricted to this key range. """ assert _IsNdbQuery(query) if filters: for f in filters: query = query.filter(ndb.FilterNode(*f)) if self.include_start: start_comparator = ">=" else: start_comparator = ">" if self.include_end: end_comparator = "<=" else: end_comparator = "<" if self.key_start: query = query.filter(ndb.FilterNode("__key__", start_comparator, self.key_start)) if self.key_end: query = query.filter(ndb.FilterNode("__key__", end_comparator, self.key_end)) return query
def filter_datastore_query(self, query, filters=None): """Add query filter to restrict to this key range. Args: query: A datastore.Query instance. filters: optional list of filters to apply to the query. Each filter is a tuple: (<property_name_as_str>, <query_operation_as_str>, <value>). User filters are applied first. Returns: The input query restricted to this key range. """ assert isinstance(query, datastore.Query) if filters: for f in filters: query.update({"%s %s" % (f[0], f[1]): f[2]}) if self.include_start: start_comparator = ">=" else: start_comparator = ">" if self.include_end: end_comparator = "<=" else: end_comparator = "<" if self.key_start: query.update({"__key__ %s" % start_comparator: self.key_start}) if self.key_end: query.update({"__key__ %s" % end_comparator: self.key_end}) return query
def make_directed_ndb_query(self, kind_class, keys_only=False): """Construct an NDB query for this key range, including the scan direction. Args: kind_class: An ndb.Model subclass. keys_only: bool, default False, use keys_only on Query? Returns: An ndb.Query instance. Raises: KeyRangeError: if self.direction is not in (KeyRange.ASC, KeyRange.DESC). """ assert issubclass(kind_class, ndb.Model) if keys_only: default_options = ndb.QueryOptions(keys_only=True) else: default_options = None query = kind_class.query(app=self._app, namespace=self.namespace, default_options=default_options) query = self.filter_ndb_query(query) if self.__get_direction(True, False): query = query.order(kind_class._key) else: query = query.order(-kind_class._key) return query
def expand(self, expression, field_type=None): if isinstance(expression, Field): out = '%s.%s' % (expression.table._tablename, expression.name) if field_type == 'string' and not expression.type in ( 'string','text','json','password'): out = 'CAST(%s AS %s)' % (out, self.types['text']) return out elif isinstance(expression, (Expression, Query)): first = expression.first second = expression.second op = expression.op optional_args = expression.optional_args or {} if not second is None: out = op(first, second, **optional_args) elif not first is None: out = op(first,**optional_args) elif isinstance(op, str): if op.endswith(';'): op=op[:-1] out = '(%s)' % op else: out = op() return out elif field_type: return str(self.represent(expression,field_type)) elif isinstance(expression,(list,tuple)): return ','.join(self.represent(item,field_type) \ for item in expression) elif isinstance(expression, bool): return '1' if expression else '0' else: return str(expression)
def tables(self, *queries): tables = set() for query in queries: if isinstance(query, Field): tables.add(query.tablename) elif isinstance(query, (Expression, Query)): if not query.first is None: tables = tables.union(self.tables(query.first)) if not query.second is None: tables = tables.union(self.tables(query.second)) return list(tables)
def NOT(self,first): nops = { self.EQ: self.NE, self.NE: self.EQ, self.LT: self.GE, self.GT: self.LE, self.LE: self.GT, self.GE: self.LT} if not isinstance(first,Query): raise SyntaxError("Not suported") nop = nops.get(first.op,None) if not nop: raise SyntaxError("Not suported %s" % first.op.__name__) first.op = nop return self.expand(first)
def _select(self,query,fields,attributes): if not isinstance(query,Query): raise SyntaxError("Not Supported") for key in set(attributes.keys())-SELECT_ARGS: raise SyntaxError('invalid select attribute: %s' % key) new_fields=[] for item in fields: if isinstance(item,SQLALL): new_fields += item._table else: new_fields.append(item) def uid(fd): return fd=='id' and '_id' or fd def get(row,fd): return fd=='id' and long(row['_id']) or row.get(fd,None) fields = new_fields tablename = self.get_table(query) fieldnames = [f.name for f in (fields or self.db[tablename])] colnames = ['%s.%s' % (tablename,k) for k in fieldnames] fields = ','.join(['%s.%s' % (tablename,uid(f)) for f in fieldnames]) fn="(function(%(t)s){if(%(query)s)emit(%(order)s,[%(fields)s]);})" %\ dict(t=tablename, query=self.expand(query), order='%s._id' % tablename, fields=fields) return fn, colnames
def update(self,tablename,query,fields): if not isinstance(query,Query): raise SyntaxError("Not Supported") if query.first.type=='id' and query.op==self.EQ: id = query.second tablename = query.first.tablename ctable = self.connection[tablename] try: doc = ctable[str(id)] for key,value in fields: doc[key.name] = self.represent(value,self.db[tablename][key.name].type) ctable.save(doc) return 1 except couchdb.http.ResourceNotFound: return 0 else: tablename = self.get_table(query) rows = self.select(query,[self.db[tablename]._id],{}) ctable = self.connection[tablename] table = self.db[tablename] for row in rows: doc = ctable[str(row.id)] for key,value in fields: doc[key.name] = self.represent(value,table[key.name].type) ctable.save(doc) return len(rows)
def count(self,query,distinct=None): if distinct: raise RuntimeError("COUNT DISTINCT not supported") if not isinstance(query,Query): raise SyntaxError("Not Supported") tablename = self.get_table(query) rows = self.select(query,[self.db[tablename]._id],{}) return len(rows)
