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_sort_rows():
from dataflows import sort_rows
f = Flow(
[
{'a': 1, 'b': 3},
{'a': 2, 'b': 3},
{'a': 3, 'b': 1},
{'a': 4, 'b': 1},
],
sort_rows(key='{b}{a}'),
)
results, _, _ = f.results()
assert list(results[0]) == [
{'a': 3, 'b': 1},
{'a': 4, 'b': 1},
{'a': 1, 'b': 3},
{'a': 2, 'b': 3},
]
characters,
set_type('age', type='number'),
sort_rows('{age:02}'),
join_with_self(
'res_1',
'{house}',
{
'the_house': {
'name': 'house'
},
'*': {
'aggregate': 'first'
},
}
),
sort_rows('{the_house}')
).results()
assert res[0] == [
{
'the_house': 'Lannister',
'first_name': 'Tyrion',
'last_name': 'Lannister',
'age': Decimal('27')
},
{
'the_house': 'Stark',
'first_name': 'Rickon',
'last_name': 'Stark',
'age': Decimal('5')
},
{
{'a': 0},
{'a': -1000000},
{'a': 1000000},
{'a': -0.1},
{'a': -0.2},
{'a': 0.2},
{'a': -1000001},
{'a': 1000001},
{'a': 6},
{'a': -10},
{'a': -0.001},
{'a': 0.001},
{'a': 1},
{'a': -1},
],
sort_rows(key='{a}'),
)
results, _, _ = f.results()
assert list(results[0]) == [
{'a': -1000001},
{'a': -1000000},
{'a': -10},
{'a': -4},
{'a': -3},
{'a': -1},
{'a': -0.2},
{'a': -0.1},
{'a': -0.001},
{'a': 0},
{'a': 0.001},
{'a': 0.1},
{'a': 0.2},
{
'avg_age': Decimal(16),
'house': 'House of Targaryen',
'max_age': Decimal(16),
'number_of_characters': 1,
'representative': 'Daenerys',
'representative_age': Decimal(16),
'last_names': [('Targaryen', 1)]
},
]
# Find youngest of each house
res, _, _ = Flow(
characters,
set_type('age', type='number'),
sort_rows('{age:02}'),
join_with_self(
'res_1',
'{house}',
{
'the_house': {
'name': 'house'
},
'*': {
'aggregate': 'first'
},
}
),
sort_rows('{the_house}')
).results()
assert res[0] == [
def test_sort_rows_datetime():
import datetime
from dataflows import sort_rows
f = Flow(
[
{'a': datetime.date(2000, 1, 3)},
{'a': datetime.date(2010, 1, 2)},
{'a': datetime.date(2020, 1, 1)},
],
sort_rows(key='{a}'),
)
results, _, _ = f.results()
assert list(results[0]) == [
{'a': datetime.date(2000, 1, 3)},
{'a': datetime.date(2010, 1, 2)},
{'a': datetime.date(2020, 1, 1)},
]
def test_sort_reverse_many_rows():
from dataflows import sort_rows
f = Flow(
({'a': i, 'b': i % 5} for i in range(1000)),
sort_rows(key='{b}{a}', reverse=True, batch_size=0),
)
results, _, _ = f.results()
results = results[0]
assert results[0:2] == [{'a': 999, 'b': 4}, {'a': 994, 'b': 4}]
assert results[998:1000] == [{'a': 5, 'b': 0}, {'a': 0, 'b': 0}]
def test_sort_rows_decimal():
from decimal import Decimal
from dataflows import sort_rows, load
f = Flow(
load('data/numbers.csv', cast_strategy=load.CAST_WITH_SCHEMA),
sort_rows(key='{a}'),
)
results, dp, _ = f.results()
assert list(results[0]) == [
{'a': Decimal('-1000')},
{'a': Decimal('-0.5')},
{'a': Decimal('-0.4')},
{'a': Decimal('0')},
{'a': Decimal('1.1')},
{'a': Decimal('2')},
{'a': Decimal('10')},
{'a': Decimal('1000')}
]
def flow(parameters):
return Flow(
load_lazy_json(parameters.get('resources')),
sort_rows(
parameters['sort-by'],
resources=parameters.get('resources'),
reverse=parameters.get('reverse')
)