Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def explain_result(self, result=None):
super(Agg, self).explain_result(result)
tuples = list(Agg._process_agg(self._result_dict['aggregations']))
assert len(tuples) > 0
self._index_names = list(tuples[0][0])
self._values = []
self._indexes = []
for t in tuples:
_, index, row = t
self._values.append(row)
if len(index) > 0:
self._indexes.append(index)
Recursively extract agg values
:param bucket: a bucket contains either sub-buckets or a bunch of aggregated values
:return: a list of tuples: (index_name, index_tuple, row)
"""
# for each agg, yield a row
row = {}
for k, v in bucket.items():
if isinstance(v, dict):
if 'buckets' in v:
for sub_bucket in v['buckets']:
if 'key_as_string' in sub_bucket:
key = sub_bucket['key_as_string']
else:
key = sub_bucket['key']
for x in Agg._process_agg(sub_bucket,
indexes + (key,),
names + (k,)):
yield x
elif 'value' in v:
row[k] = v['value']
elif 'values' in v: # percentiles
row = v['values']
else:
row.update(v) # stats
else:
if k == 'doc_count': # count docs
row['doc_count'] = v
if len(row) > 0:
yield (names, indexes, row)