def agg_iter(self, lower_limit=None, upper_limit=None):
lower_limit = lower_limit or self.get_bookmark().isoformat()<EOL>upper_limit = upper_limit or (<EOL>datetime.datetime.utcnow().replace(microsecond=<NUM_LIT:0>).isoformat())<EOL>aggregation_data = {}<EOL>self.agg_query = Search(using=self.client,<EOL>index=self.event_index).filter('<STR_LIT>', timestamp={<EOL>'<STR_LIT>': self._format_range_dt(lower_limit),<EOL>'<STR_LIT>': self._format_range_dt(upper_limit)})<EOL>for modifier in self.query_modifiers:<EOL><INDENT>self.agg_query = modifier(self.agg_query)<EOL><DEDENT>hist = self.agg_query.aggs.bucket(<EOL>'<STR_LIT>',<EOL>'<STR_LIT>',<EOL>field='<STR_LIT>',<EOL>interval=self.aggregation_interval<EOL>)<EOL>terms = hist.bucket(<EOL>'<STR_LIT>', '<STR_LIT>', field=self.aggregation_field, size=<NUM_LIT:0><EOL>)<EOL>top = terms.metric(<EOL>'<STR_LIT>', '<STR_LIT>', size=<NUM_LIT:1>, sort={'<STR_LIT>': '<STR_LIT>'}<EOL>)<EOL>for dst, (metric, src, opts) in self.metric_aggregation_fields.items():<EOL><INDENT>terms.metric(dst, metric, field=src, **opts)<EOL><DEDENT>results = self.agg_query.execute()<EOL>index_name = None<EOL>for interval in results.aggregations['<STR_LIT>'].buckets:<EOL><INDENT>interval_date = datetime.datetime.strptime(<EOL>interval['<STR_LIT>'], '<STR_LIT>')<EOL>for aggregation in interval['<STR_LIT>'].buckets:<EOL><INDENT>aggregation_data['<STR_LIT>'] = interval_date.isoformat()<EOL>aggregation_data[self.aggregation_field] = aggregation['<STR_LIT:key>']<EOL>aggregation_data['<STR_LIT:count>'] = aggregation['<STR_LIT>']<EOL>if self.metric_aggregation_fields:<EOL><INDENT>for f in self.metric_aggregation_fields:<EOL><INDENT>aggregation_data[f] = aggregation[f]['<STR_LIT:value>']<EOL><DEDENT><DEDENT>doc = aggregation.top_hit.hits.hits[<NUM_LIT:0>]['<STR_LIT>']<EOL>for destination, source in self.copy_fields.items():<EOL><INDENT>if isinstance(source, six.string_types):<EOL><INDENT>aggregation_data[destination] = doc[source]<EOL><DEDENT>else:<EOL><INDENT>aggregation_data[destination] = source(<EOL>doc,<EOL>aggregation_data<EOL>)<EOL><DEDENT><DEDENT>index_name = '<STR_LIT>'.format(self.event,<EOL>interval_date.strftime(<EOL>self.index_name_suffix))<EOL>self.indices.add(index_name)<EOL>yield dict(_id='<STR_LIT>'.<EOL>format(aggregation['<STR_LIT:key>'],<EOL>interval_date.strftime(<EOL>self.doc_id_suffix)),<EOL>_index=index_name,<EOL>_type=self.aggregation_doc_type,<EOL>_source=aggregation_data)<EOL><DEDENT><DEDENT>self.last_index_written = index_name<EOL>
Aggregate and return dictionary to be indexed in ES.