How to use the dataflows.sort_rows function in dataflows

To help you get started, we’ve selected a few dataflows examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github datahq / dataflows / tests / test_lib.py View on Github external
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},
    ]
github datahq / dataflows / tests / test_lib.py View on Github external
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')
        },
        {
github datahq / dataflows / tests / test_lib.py View on Github external
{'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},
github datahq / dataflows / tests / test_lib.py View on Github external
{
            '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] == [
github datahq / dataflows / tests / test_lib.py View on Github external
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)},
    ]
github datahq / dataflows / tests / test_lib.py View on Github external
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}]
github datahq / dataflows / tests / test_lib.py View on Github external
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')}
    ]
github frictionlessdata / datapackage-pipelines / datapackage_pipelines / lib / sort.py View on Github external
def flow(parameters):
    return Flow(
        load_lazy_json(parameters.get('resources')),
        sort_rows(
            parameters['sort-by'],
            resources=parameters.get('resources'),
            reverse=parameters.get('reverse')
        )