Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_select_explain_result(self):
select = Select()
select._result_dict = create_hits()
select.explain_result()
print(select)
print(repr(select))
self.assertIsNotNone(select.result)
self.assertEqual(len(select), 3)
def test_collect(self):
query = Select()
query._values = [{'a': 1}, {'a': 2}]
self.assertEqual(query.collect(), [Row(a=1), Row(a=2)])
def test_select_explain_result(self):
select = Select()
select._result_dict = create_hits()
select.explain_result()
print(select)
print(repr(select))
self.assertIsNotNone(select.result)
self.assertEqual(len(select), 3)
def __init__(self):
super(Select, self).__init__()
def from_dict(d):
query = Select()
query.explain_result(d)
return query
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 explain_result(self, result=None):
super(Select, self).explain_result(result)
rows = []
for hit in self._result_dict['hits']['hits']:
row = {}
for k in hit.keys():
if k == '_source':
solved_fields = self.resolve_fields(hit['_source'])
row.update(solved_fields)
elif k.startswith('_'):
row[k] = hit[k]
rows.append(row)
self._values = rows
def result_as_tabular(self, cols, n, truncate=20):
b = six.StringIO()
widths = []
tavnit = '|'
separator = '+'
for col in cols:
maxlen = len(col)
for kv in self.result[:n]:
if col in kv:
s = Select._stringfy_value(kv[col])
else:
s = '(NULL)'
if len(s) > maxlen:
maxlen = len(s)
widths.append(min(maxlen, truncate))
for w in widths:
tavnit += ' %-' + '%ss |' % (w,)
separator += '-' * w + '--+'
b.write(separator + '\n')
b.write(tavnit % tuple(cols) + '\n')
b.write(separator + '\n')
for kv in self.result[:n]:
row = []
for col in cols: