Produits Entr'ouvert » Passerelle

DELIMITER = '__'

class InvalidOperatorError(Exception):

    pass

compare_str = cmp

def is_int(value):

    try:

        int(value)

        return True

    except (ValueError, TypeError):

        return False

class Lookup(object):

    def contains(self, key, value):

        return lambda x: value in x[key]

    def icontains(self, key, value):

        return lambda x: value.lower() in x[key].lower()

    def gt(self, key, value):

        return lambda x: int(x[key]) > int(value)

    def igt(self, key, value):

        return lambda x: compare_str(x[key].lower(), value.lower()) > 0

    def ge(self, key, value):

        return lambda x: int(x[key]) >= int(value)

    def ige(self, key, value):

        return lambda x: compare_str(x[key].lower(), value.lower()) >= 0

    def lt(self, key, value):

        return lambda x: int(x[key]) < int(value)

    def ilt(self, key, value):

        return lambda x: compare_str(x[key].lower(), value.lower()) < 0

    def le(self, key, value):

        return lambda x: int(x[key]) <= int(value)

    def ile(self, key, value):

        return lambda x: compare_str(x[key].lower(), value.lower()) <= 0

    def eq(self, key, value):

        if is_int(value):

            return lambda x: int(value) == int(x[key])

        return lambda x: value == x[key]

    def ieq(self, key, value):

        return lambda x: value.lower() == x[key].lower()

    def ne(self, key, value):

        if is_int(value):

            return lambda x: int(value) != int(x[key])

        return lambda x: value != x[key]

    def ine(self, key, value):

        return lambda x: value.lower() != x[key].lower()

def get_lookup(operator, key, value):

    try:

        return getattr(Lookup(), operator)(key, value)

    except (AttributeError,):

        raise InvalidOperatorError('%s is not a valid operator' % operator)

import lookups

    def get_data(self, filters=None):

        titles = [t.strip() for t in self.columns_keynames.split(',')]

        # validate filters (appropriate columns must exist)

        if filters:

            for filter_key in filters.keys():

                if not filter_key.split(lookups.DELIMITER)[0] in titles:

                    del filters[filter_key]

        rows = self.get_cached_rows()

        data = []

        # build a generator of all filters

        def filters_generator(filters, titles):

            if not filters:

                return

            for key, value in filters.items():

                try:

                    key, op = key.split(lookups.DELIMITER)

                except (ValueError,):

                    op = 'eq'

                yield lookups.get_lookup(op, key, value)

        # apply filters to data

        def super_filter(filters, data):

            for f in filters:

                data = itertools.ifilter(f, data)

            return data

        data = list(super_filter(

            filters_generator(filters, titles), rows

        ))

        return data

    def _filters_builder(self, request):

        filters = {}

        obj = self.get_object()

        params = request.GET

        case_insensitive = 'case-insensitive' in params

        query = params.get('q', None)

        if query:

            if case_insensitive:

                filters['text__icontains'] = query.lower()

            else:

                filters['text__contains'] = query

        # builds filters according to csv file header

        for column_title in [t.strip() for t in obj.columns_keynames.split(',') if t]:

            match = filter(

                (lambda ct: lambda x: x.startswith(ct))(column_title), params.keys()

            )

            for key in match:

                if case_insensitive:

                    filters[key + '__ieq'] = params[key].lower()

                else:

                    filters[key] = params[key]

        if 'text' in filters:

            if case_insensitive:

                filters['text__ieq'] = filters['text'].lower()

            else:

                filters['text__eq'] = filters['text']

            filters.pop('text')

        return filters

        obj = self.get_object()

        filters = self._filters_builder(request)

        return {'data': obj.get_data(filters)}

def test_skipped_header_data():

    csv = CsvDataSource.objects.create(csv_file=File(StringIO(get_file_content('data.csv')), 'data.csv'),

                                       columns_keynames=',id,,text,',

                                       skip_header=True)

    result = csv.get_data({'text': 'Eliot'})

    assert len(result) == 0

def test_advanced_filters(client, setup, filetype):

    csvdata, url = setup(filename=filetype, data=get_file_content(filetype))

    filters = {'id__gt':20, 'id__lt': 40}

    resp = client.get(url, filters)

    result = parse_response(resp)

    assert len(result) == 3

    for stuff in result:

        assert stuff['id'] in ('22', '36', '38')

def test_advanced_filters_combo(client, setup, filetype):

    csvdata, url = setup(filename=filetype, data=get_file_content(filetype))

    filters = {

        'id__ge': '20',

        'id__lt': '40',

        'fam__gt': '234',

        'fam__le': '235',

        'fname__icontains': 'Sandra'

    }

    resp = client.get(url, filters)

    result = parse_response(resp)

    assert len(result) == 1

    assert result[0]['id'] == '22'

    assert result[0]['lname'] == 'MARTIN'

def test_unknown_operator(client, setup, filetype):

    csvdata, url = setup(filename=filetype, data=get_file_content(filetype))

    filters = {'id__whatever': '25', 'fname__icontains':'Eliot'}

    resp = client.get(url, filters)

    result = json.loads(resp.content)

    assert result['err'] == 1

    assert result['err_class'] == 'passerelle.apps.csvdatasource.lookups.InvalidOperatorError'

    assert result['err_desc'] == 'whatever is not a valid operator'

    filters = {'id__gt': '20', 'id__lt': '40', 'fname__icontains': 'Sandra'}

-