@tasklets.tasklet<EOL><INDENT>def run_to_queue(self, queue, conn, options=None):<DEDENT>
if options is None:<EOL><INDENT>offset = None<EOL>limit = None<EOL>keys_only = None<EOL><DEDENT>else:<EOL><INDENT>offset = options.offset<EOL>limit = options.limit<EOL>keys_only = options.keys_only<EOL>if (options.start_cursor or options.end_cursor or<EOL>options.produce_cursors):<EOL><INDENT>names = set()<EOL>if self.__orders is not None:<EOL><INDENT>names = self.__orders._get_prop_names()<EOL><DEDENT>if '<STR_LIT>' not in names:<EOL><INDENT>raise datastore_errors.BadArgumentError(<EOL>'<STR_LIT>')<EOL><DEDENT><DEDENT><DEDENT>modifiers = {}<EOL>if offset:<EOL><INDENT>modifiers['<STR_LIT>'] = None<EOL>if limit is not None:<EOL><INDENT>modifiers['<STR_LIT>'] = min(_MAX_LIMIT, offset + limit)<EOL><DEDENT><DEDENT>if keys_only and self.__orders is not None:<EOL><INDENT>modifiers['<STR_LIT>'] = None<EOL><DEDENT>if modifiers:<EOL><INDENT>options = QueryOptions(config=options, **modifiers)<EOL><DEDENT>if offset is None:<EOL><INDENT>offset = <NUM_LIT:0><EOL><DEDENT>if limit is None:<EOL><INDENT>limit = _MAX_LIMIT<EOL><DEDENT>if self.__orders is None:<EOL><INDENT>keys_seen = set()<EOL>for subq in self.__subqueries:<EOL><INDENT>if limit <= <NUM_LIT:0>:<EOL><INDENT>break<EOL><DEDENT>subit = tasklets.SerialQueueFuture('<STR_LIT>')<EOL>subq.run_to_queue(subit, conn, options=options)<EOL>while limit > <NUM_LIT:0>:<EOL><INDENT>try:<EOL><INDENT>batch, index, result = yield subit.getq()<EOL><DEDENT>except EOFError:<EOL><INDENT>break<EOL><DEDENT>if keys_only:<EOL><INDENT>key = result<EOL><DEDENT>else:<EOL><INDENT>key = result._key<EOL><DEDENT>if key not in keys_seen:<EOL><INDENT>keys_seen.add(key)<EOL>if offset > <NUM_LIT:0>:<EOL><INDENT>offset -= <NUM_LIT:1><EOL><DEDENT>else:<EOL><INDENT>limit -= <NUM_LIT:1><EOL>queue.putq((None, None, result))<EOL><DEDENT><DEDENT><DEDENT><DEDENT>queue.complete()<EOL>return<EOL><DEDENT>with conn.adapter:<EOL><INDENT>todo = []  <EOL>for subq in self.__subqueries:<EOL><INDENT>dsquery = subq._get_query(conn)<EOL>subit = tasklets.SerialQueueFuture('<STR_LIT>')<EOL>subq.run_to_queue(subit, conn, options=options, dsquery=dsquery)<EOL>todo.append((subit, dsquery))<EOL><DEDENT>state = []  <EOL>for subit, dsquery in todo:<EOL><INDENT>try:<EOL><INDENT>thing = yield subit.getq()<EOL><DEDENT>except EOFError:<EOL><INDENT>continue<EOL><DEDENT>else:<EOL><INDENT>state.append(_SubQueryIteratorState(thing, subit, dsquery,<EOL>self.__orders))<EOL><DEDENT><DEDENT>heapq.heapify(state)<EOL>keys_seen = set()<EOL>while state and limit > <NUM_LIT:0>:<EOL><INDENT>item = heapq.heappop(state)<EOL>batch = item.batch<EOL>index = item.index<EOL>entity = item.entity<EOL>key = entity._key<EOL>if key not in keys_seen:<EOL><INDENT>keys_seen.add(key)<EOL>if offset > <NUM_LIT:0>:<EOL><INDENT>offset -= <NUM_LIT:1><EOL><DEDENT>else:<EOL><INDENT>limit -= <NUM_LIT:1><EOL>if keys_only:<EOL><INDENT>queue.putq((batch, index, key))<EOL><DEDENT>else:<EOL><INDENT>queue.putq((batch, index, entity))<EOL><DEDENT><DEDENT><DEDENT>subit = item.iterator<EOL>try:<EOL><INDENT>batch, index, entity = yield subit.getq()<EOL><DEDENT>except EOFError:<EOL><INDENT>pass<EOL><DEDENT>else:<EOL><INDENT>item.batch = batch<EOL>item.index = index<EOL>item.entity = entity<EOL>heapq.heappush(state, item)<EOL><DEDENT><DEDENT>queue.complete()<EOL><DEDENT>
Run this query, putting entities into the given queue.