Produits Entr'ouvert » Passerelle

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import csv

import datetime

import os

import re

import tempfile

from collections import OrderedDict

import pytz

import six

from django.conf import settings

from django.contrib.postgres.fields import JSONField

from django.core.exceptions import ValidationError

from django.db import models, transaction

from django.urls import reverse

from django.utils.encoding import force_str, force_text, smart_text

from django.utils.timezone import make_aware, now

from django.utils.translation import ugettext_lazy as _

from pyexcel_ods import get_data as get_data_ods

from pyexcel_xls import get_data as get_data_xls

from passerelle.base.models import BaseResource

from passerelle.utils import batch

from passerelle.utils.api import endpoint

from passerelle.utils.conversion import normalize

from passerelle.utils.jsonresponse import APIError

identifier_re = re.compile(r"^[^\d\W]\w*\Z", re.UNICODE)

code_cache = OrderedDict()

def get_code(expr):

    # limit size of code cache to 1024

            self.cache_data()

            for data in self.get_cached_rows(initial=False):

                yield data

    @property

    def titles(self):

        return [smart_text(t.strip()) for t in self.columns_keynames.split(',')]

    @endpoint(perm='can_access', methods=['get'], name='data')

    def data(self, request, **kwargs):

        params = request.GET

        filters = []

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

            if column_title in params.keys():

                query_value = request.GET.get(column_title, '')

                if 'case-insensitive' in params:

  {% if object|can_edit:request.user %}<a href="{% url 'csv-download' connector_slug=object.slug %}">{{object.csv_file}}</a>

  — {% trans "added on" %} {{object.csv_file_datetime}}

  {% else %}{{object.csv_file}}{% endif %}

</p>

{% endblock %}

{% block endpoints %}

<ul class="endpoints">

{% url 'generic-endpoint' connector='csvdatasource' slug=object.slug endpoint='data' as csvdatasource_data_url %}

<li class="get-method">

  <div class="description"><span class="description--label">{% trans "Returning all file lines:" %}</span>

  <a class="example-url" href="{{ csvdatasource_data_url }}">{{ csvdatasource_data_url }}</a></div>

</li>

<li class="get-method">

  <div class="description"><span class="description--label">{% trans "Returning lines containing 'abc' in 'text' column (case insensitive):" %}</span>

  <a class="example-url" href="{{ csvdatasource_data_url }}?q=abc">{{ csvdatasource_data_url }}?q=abc</a></div>

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import datetime

import os

import time

import uuid

from posix import stat_result

from stat import ST_MTIME

import mock

import pytest

import six

import webtest

from django.contrib.contenttypes.models import ContentType

from django.core.files import File

from django.core.management import call_command

from django.test import Client, override_settings

from django.urls import reverse

from django.utils.encoding import force_bytes, force_str, force_text

from django.utils.six import StringIO

from django.utils.six.moves.urllib.parse import urlencode

from django.utils.timezone import now

from test_manager import login

from passerelle.base.models import AccessRight, ApiUser

from passerelle.compat import json_loads

data = """121;69981;DELANOUE;Eliot;H

525;6;DANIEL WILLIAMS;Shanone;F

253;67742;MARTIN;Sandra;F

511;38142;MARCHETTI;Lucie;F

235;22;MARTIN;Sandra;F

    csv.refresh_from_db()

    assert list(csv.get_rows()) == []

    call_command('change-csv', 'test', os.path.join(TEST_BASE_DIR, 'data.ods'), sheet_name='Feuille2')

    csv.refresh_from_db()

    assert list(csv.get_rows()) != []

@pytest.mark.parametrize(

    'payload,expected',

    [

        ({}, 20),

        ({'limit': 10}, 10),

        ({'limit': 10, 'offset': 0}, 10),

        ({'limit': 10, 'offset': 15}, 5),

        ({'limit': 10, 'offset': 42}, 0),