Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _execute(self):
if self._client is None:
raise _unbound_index_err
res_dict = self._client.post(data=self._build_query())
if self._aggregation is None and self._groupby is None:
query = Select()
query.explain_result(res_dict)
else:
query = Agg.from_dict(res_dict)
return query
def aggregate(self, *args):
query = self._build_query(aggs=args, filter=self._filter, size=0)
return Agg.from_dict(self._client.post(data=query))