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_collect(self):
query = Select()
query._values = [{'a': 1}, {'a': 2}]
self.assertEqual(query.collect(), [Row(a=1), Row(a=2)])
def test_row(self):
row = Row(a=1, b='你好,世界')
print(repr(row))
self.assertEqual(row['a'], 1)
self.assertEqual(row['b'], '你好,世界')
self.assertEqual(row.as_dict(), {'a': 1, 'b': '你好,世界'})
def __getitem__(self, name):
try:
idx = self._fields.index(name)
return super(Row, self).__getitem__(idx)
except IndexError:
raise KeyError(name)
except ValueError:
raise ValueError(name)
def __repr__(self):
return 'Row(' + ','.join(
['{0}={1}'.format(k, Row._stringfy(v)) for k, v in zip(self._fields, tuple(self))]) + ')'
def collect(self):
"""
Returns all the records as a list of Row.
:return: list of :class:`Row `
>>> df.collect()
[Row(age=2, name='Alice'), Row(age=5, name='Bob')]
"""
query = self._execute()
return [Row(**v) for v in query.result]